src/lib/gscprts/rts_main.c

   1 /* ------------------------------------------------
   2  Copyright 2014 AT&T Intellectual Property
   3  Licensed under the Apache License, Version 2.0 (the "License");
   4  you may not use this file except in compliance with the License.
   5  You may obtain a copy of the License at
   6
   7  http://www.apache.org/licenses/LICENSE-2.0
   8
   9  Unless required by applicable law or agreed to in writing, software
  10  distributed under the License is distributed on an "AS IS" BASIS,
  11  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  See the License for the specific language governing permissions and
  13  limitations under the License.
  14  ------------------------------------------- */
  15
  16 #include "gsconfig.h"
  17 #include "gstypes.h"
  18 #include "gshub.h"
  19
  20 #include "lapp.h"
  21 #include "fta.h"
  22 #include "lfta/rts.h"
  23
  24 #include "stdio.h"
  25 #include "stdlib.h"
  26 #include <sys/types.h>
  27 #include <unistd.h>
  28
  29 #include <sys/ipc.h>
  30 #include <sys/shm.h>
  31 #include <fcntl.h>
  32 #include <errno.h>
  33 #include <sys/socket.h>
  34
  35 gs_retval_t main_csv(gs_int32_t devicenum, gs_sp_t device, gs_int32_t mapcnt, gs_sp_t map[]);
  36 gs_retval_t main_gdat(gs_int32_t devicenum, gs_sp_t device, gs_int32_t mapcnt, gs_sp_t map[]);
  37 gs_retval_t main_dproto(gs_int32_t devicenum, gs_sp_t device, gs_int32_t mapcnt, gs_sp_t map[]);
  38 gs_retval_t main_kafka(gs_int32_t devicenum, gs_sp_t device, gs_int32_t mapcnt, gs_sp_t map[]);
  39
  40 int main (int argc, char* argv[]) {
  41     gs_int32_t pid;
  42     gs_int32_t x;
  43     gs_int32_t y;
  44     gs_sp_t* device;
  45     gs_int32_t devcnt=0;
  46     gs_sp_t* mappings;
  47     gs_int32_t mapcnt=0;
  48     gs_sp_t* lmap;
  49     gs_int32_t lmapcnt=0;
  50     FILE * cfg_file;
  51     gs_int32_t tip1,tip2,tip3,tip4;
  52     endpoint gshub;
  53     gs_sp_t instance_name;
  54     gs_csp_t kafka;
  55
  56
  57         gsopenlog(argv[0]);
  58
  59     if (setpgid(0,0)<0) {
  60         gslog(LOG_EMERG,"Could not set process group id of rts");
  61         exit(1);
  62     }
  63
  64     if (argc<4) {
  65                 gslog(LOG_EMERG,"Wrong arguments at startup");
  66         exit(1);
  67     }
  68
  69     /* allocate more than enough for each array */
  70     if ((device=(gs_sp_t*)malloc(sizeof(gs_sp_t*) * argc))==0) {
  71         gslog(LOG_EMERG,"malloc error");
  72         exit(1);
  73     }
  74     if ((mappings=(gs_sp_t*)malloc(sizeof(gs_sp_t*) * argc))==0) {
  75         gslog(LOG_EMERG,"malloc error");
  76         exit(1);
  77     }
  78     if ((lmap=(gs_sp_t*)malloc(sizeof(gs_sp_t*) * argc))==0) {
  79         gslog(LOG_EMERG,"malloc error");
  80         exit(1);
  81     }
  82
  83     /* parse the arguments */
  84
  85     if ((sscanf(argv[1],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(gshub.port))!=5)) {
  86         gslog(LOG_EMERG,"HUB IP NOT DEFINED");
  87         exit(1);
  88     }
  89     gshub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
  90     gshub.port=htons(gshub.port);
  91     instance_name=strdup(argv[2]);
  92     if (set_hub(gshub)!=0) {
  93         gslog(LOG_EMERG,"Could not set hub");
  94         exit(1);
  95     }
  96     if (set_instance_name(instance_name)!=0) {
  97         gslog(LOG_EMERG,"Could not set instance name");
  98         exit(1);
  99     }
 100
 101     for(x=3;x<argc;x++) {
 102         if (strncmp("-D",argv[x],2)==0) {
 103             /* macro definition */
 104             y=2;
 105             while((y<(strlen(argv[x])-1))&&(argv[x][y]!='=')) y++;
 106
 107
 108             if (y<(strlen(argv[x])-1)) {
 109                 /* if that is not true the define is empty and
 110                  we ignore it otherwise we set the = to 0 to
 111                  make two strings out of it*/
 112                 argv[x][y]=0;
 113                 mappings[mapcnt]=&(argv[x][2]);
 114                 mapcnt++;
 115             }
 116         } else {
 117             /* device definition */
 118             device[devcnt]=argv[x];
 119             devcnt++;
 120         }
 121     }
 122
 123     if (devcnt==0) {
 124         gslog(LOG_EMERG,"at least one device has to be specified");
 125         exit(1);
 126     }
 127
 128
 129     /* now startup all the device dependend processes. */
 130
 131     for (x=0;x<devcnt;x++) {
 132         if ((pid=fork())==-1) {
 133             gslog(LOG_EMERG,"fork error");
 134             exit(1);
 135         }
 136
 137         if (pid==0) {
 138             gs_csp_t interfacetype;
 139             /* wait for clearinghouse to finish startup */
 140             sleep(2);
 141             for(y=0;y<mapcnt;y++) {
 142                 if (strcmp(device[x],&mappings[y][strlen(mappings[y])+1])==0) {
 143                     /* point to the second string */
 144                     lmap[lmapcnt]=mappings[y];
 145                     lmapcnt++;
 146                 }
 147             }
 148             /* the devicename always matches */
 149             lmap[lmapcnt]=device[x];
 150             lmapcnt++;
 151
 152             if ((interfacetype=get_iface_properties(device[x],(gs_sp_t)"interfacetype"))==0) {
 153                 gslog(LOG_EMERG,"Interface Type not configured but required");
 154                 exit(1);
 155             }
 156
 157             if ((kafka = get_iface_properties(device[x],(gs_sp_t)"kafka")) && !strcmp(kafka,"TRUE")) {
 158                 main_kafka(x,device[x],lmapcnt,lmap);
 159             } else if (strcmp(interfacetype,"CSV")==0) {
 160                 main_csv(x,device[x],lmapcnt,lmap);
 161             } else if (strncmp(interfacetype,"GDAT",4)==0) {
 162                 main_gdat(x,device[x],lmapcnt,lmap);
 163                         } else if (strncmp(interfacetype,"PROTO",5)==0) {
 164                 main_dproto(x,device[x],lmapcnt,lmap);
 165             } else {
 166                     fprintf(stderr,"interface type %s not recognized\n",interfacetype);
 167                     gslog(LOG_ERR,"UNKNOWN InterfaceType\n");
 168                     exit(0);
 169             }
 170
 171             /* should never return */
 172             gslog(LOG_EMERG,"lfta init returned");
 173             exit(1);
 174         }
 175     }
 176
 177     /* initalize host_lib */
 178
 179     gslog(LOG_INFO,"Init host lib in clearinghouse");
 180
 181     if (hostlib_init(CLEARINGHOUSE,0,DEFAULTDEV,0,0)<0) {
 182         gslog(LOG_EMERG,"%s::error:could not initiate host lib for clearinghouse\n",
 183               argv[0]);
 184         exit(7);
 185     }
 186
 187
 188
 189     /* start processing messages should never return*/
 190     if (fta_start_service(-1)<0) {
 191         gslog(LOG_EMERG,"error in processing the msg queue");
 192         exit(9);
 193     }
 194     gslog(LOG_EMERG,"%s::error:start service returned");
 195     return 0;
 196 }
 197