src/lib/gscprts/rts_csv.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 #include <time.h>
  16 #include <stdlib.h>
  17 #include <string.h>
  18 #include <unistd.h>
  19 #include <fcntl.h>
  20 #include <sys/time.h>
  21 #include <sys/stat.h>
  22 #include <sys/types.h>
  23 #include <sys/socket.h>
  24 #include <netinet/in.h>
  25 #include "errno.h"
  26
  27 #include "gsconfig.h"
  28 #include "gshub.h"
  29 #include "gstypes.h"
  30 #include "lapp.h"
  31 #include "fta.h"
  32 #include "stdio.h"
  33 #include "stdlib.h"
  34 #include "packet.h"
  35 #include "schemaparser.h"
  36 #include "lfta/rts.h"
  37
  38 void rts_fta_process_packet(struct packet * p);
  39 void rts_fta_done();
  40 void fta_init(gs_sp_t device);
  41
  42
  43 #define CSVMAXLINE 1000000
  44
  45 static FILE *pd;
  46 static int listensockfd=-1;
  47 static int fd=-1;
  48 static struct packet cur_packet;
  49 static gs_sp_t name;
  50 static gs_uint8_t line[CSVMAXLINE];
  51 static gs_uint32_t lineend=0;
  52 static gs_uint8_t csvdel=',';
  53 static gs_uint32_t verbose=0;
  54 static gs_uint32_t startupdelay=0;
  55 static gs_uint32_t singlefile=0;
  56 static gs_uint32_t gshub=0;
  57 static int socket_desc=0;
  58
  59 static void csv_replay_check_messages() {
  60     if (fta_start_service(0)<0) {
  61         print_error("Error:in processing the msg queue for a replay file");
  62         exit(9);
  63     }
  64 }
  65
  66 static gs_uint32_t gs_read_line(gs_uint8_t * buffer, gs_uint32_t length){
  67     gs_uint32_t used=0;
  68     gs_uint32_t cur;
  69     fd_set socket_rset;
  70     fd_set socket_eset;
  71     struct timeval socket_timeout;
  72     int retval;
  73
  74     FD_ZERO(&socket_rset);
  75     FD_SET(socket_desc,&socket_rset);
  76     FD_ZERO(&socket_eset);
  77     FD_SET(socket_desc,&socket_eset);
  78     // timeout in one millisecon
  79     socket_timeout.tv_sec=0;
  80     socket_timeout.tv_usec=1000;
  81
  82     if ((retval=select(socket_desc+1,&socket_rset,0,&socket_eset,&socket_timeout))<=0) {
  83         if (retval==0) {
  84             // caught a timeout
  85             return -1;
  86         }
  87         return -2;
  88     }
  89
  90     while((used < (length-1)) && ((used==0)|| (buffer[used-1]!='\n'))) {
  91         if ((cur=read(socket_desc,&(buffer[used]),1))<=0) {
  92             print_error("ERROR:could not read data from gdat stream");
  93             return -2;
  94         }
  95         used+=cur;
  96     }
  97         buffer[used]=0;
  98         return 0;
  99 }
 100
 101 static void init_socket() {
 102         endpoint gshub;
 103         endpoint srcinfo;
 104         struct sockaddr_in server;
 105     gs_int32_t parserversion;
 106     gs_uint32_t schemalen;
 107     static char * asciischema=0;
 108         gs_int8_t buf[1024];
 109
 110         if (get_hub(&gshub)!=0) {
 111                 print_error("ERROR:could not find gshub for data source");
 112                 exit(0);
 113         }
 114
 115         if (get_streamsource(gshub,name,&srcinfo,1) !=0) {
 116                 print_error("ERROR:could not find data source for stream\n");
 117                 exit(0);
 118         }
 119
 120         socket_desc = socket(AF_INET , SOCK_STREAM , 0);
 121     if (socket_desc == -1)
 122     {
 123         print_error("ERROR:could not create socket for data stream");
 124                 exit(0);
 125     }
 126         server.sin_addr.s_addr = srcinfo.ip;
 127     server.sin_family = AF_INET;
 128     server.sin_port = srcinfo.port;
 129
 130         if (connect(socket_desc , (struct sockaddr *)&server , sizeof(server)) < 0)
 131     {
 132                 print_error("ERROR: could not open connection to data source");
 133                 exit(0);
 134         }
 135
 136 }
 137
 138 static void next_file() {
 139         struct stat s;
 140         if (verbose) {
 141                 fprintf(stderr,"Opening %s\n",name);
 142         }
 143         if (singlefile==0) {
 144                 while (lstat(name,&s)!=0) {
 145                         if (errno!=ENOENT) {
 146                                 print_error("csv::lstat unexpected return value");
 147                                 exit(10);
 148                         }
 149                         csv_replay_check_messages();
 150                         usleep(10000);
 151                 }
 152                 if  (pd!=0) {
 153                         fclose(pd);
 154                 }
 155         }
 156     if ((pd=fopen(name,"r"))==0) {
 157         print_error("csv::open failed ");
 158         exit(10);
 159     }
 160         if (singlefile==0) {
 161                 unlink(name);
 162         }
 163 }
 164
 165
 166 static gs_retval_t csv_replay_init(gs_sp_t device)
 167 {
 168     gs_sp_t  verbosetmp;
 169     gs_sp_t  delaytmp;
 170     gs_sp_t  gshubtmp;
 171     gs_sp_t  tempdel;
 172     gs_sp_t  singlefiletmp;
 173
 174     if ((name=get_iface_properties(device,"filename"))==0) {
 175                 print_error("csv_init::No CSV \"Filename\" defined");
 176                 exit(0);
 177         }
 178     tempdel=get_iface_properties(device,"csvseparator");
 179     if (tempdel != 0 ) {
 180         csvdel=(gs_uint8_t) tempdel[0];
 181     }
 182
 183     if ((verbosetmp=get_iface_properties(device,"verbose"))!=0) {
 184         if (strncmp(verbosetmp,"TRUE",4)==0) {
 185             verbose=1;
 186             fprintf(stderr,"VERBOSE ENABLED\n");
 187         } else {
 188             fprintf(stderr,"VERBOSE DISABLED\n");
 189         }
 190     }
 191     if ((singlefiletmp=get_iface_properties(device,"singlefile"))!=0) {
 192         if (strncmp(singlefiletmp,"TRUE",4)==0) {
 193             singlefile=1;
 194             if (verbose)
 195                 fprintf(stderr,"SINGLEFILE ENABLED\n");
 196         } else {
 197             if (verbose)
 198                 fprintf(stderr,"SINGLEFILE DISABLED\n");
 199         }
 200     }
 201
 202     if ((delaytmp=get_iface_properties(device,"startupdelay"))!=0) {
 203         if (verbose) {
 204             fprintf(stderr,"Startup delay of %u seconds\n",atoi(get_iface_properties(device,"startupdelay")));
 205         }
 206         startupdelay=atoi(get_iface_properties(device,"startupdelay"));
 207     }
 208     if ((gshubtmp=get_iface_properties(device,"gshub"))!=0) {
 209         if (verbose) {
 210             fprintf(stderr,"GDAT format using gshub\n");
 211         }
 212         gshub=1;
 213     }
 214
 215     cur_packet.ptype=PTYPE_CSV;
 216     return 0;
 217 }
 218
 219 static gs_retval_t csv_read_socket()
 220 {
 221     gs_uint32_t i;
 222     gs_uint32_t p;
 223         gs_uint32_t x;
 224         gs_int32_t r;
 225         gs_retval_t ret;
 226     gs_uint32_t done;
 227
 228     if ((ret=gs_read_line(line,CSVMAXLINE-1))<0) { return ret;}
 229         cur_packet.systemTime=time(0);
 230         p=0;
 231         i=0;
 232         done=0;
 233         while((done==0)&&(i<CSVMAXLINE)&&(line[i]!=0)&&(p<CSVELEMENTS)){
 234             cur_packet.record.csv.fields[p]=&line[i];
 235             p++;
 236             while((line[i] != 0) && (line[i] != csvdel)) {
 237                 i++;
 238             }
 239             if (line[i]==0) done=1;
 240             line[i]=0;
 241             i++;
 242         }
 243         cur_packet.record.csv.numberfields=p;
 244         //fprintf(stderr,"XX,%s,%s,%u,%u\n",cur_packet.record.csv.fields[0],cur_packet.record.csv.fields[1],x,lineend);
 245         rts_fta_process_packet(&cur_packet);
 246         if (lineend>x+1)  {
 247             memcpy(&(line[0]),&(line[x+1]),lineend-x-1);
 248             lineend=lineend-x-1;
 249         } else {
 250             lineend=0;
 251         }
 252         line[lineend]=0;
 253         return 0;
 254     }
 255
 256     static void csv_read_tuple()
 257     {
 258         gs_uint32_t i=0;
 259         gs_uint32_t p=0;
 260         gs_uint32_t flen=0;
 261         if (pd==0) next_file();
 262         while(fgets((char *)&(line[0]),CSVMAXLINE,pd)==0) {
 263             if (singlefile==1) {
 264                 if(verbose) {
 265                     fprintf(stderr,"SINGLEFILE PROCESSING DONE!\n");
 266                 }
 267                 rts_fta_done();
 268                 if (verbose) {
 269                     fprintf(stderr,"RTS SAYS BY\n");
 270                 }
 271                 while(1==1) sleep(1);
 272             } else {
 273                 next_file();
 274             }
 275         }
 276         cur_packet.systemTime=time(0);
 277         while((i<CSVMAXLINE)&&(line[i]!=0)&&(p<CSVELEMENTS)){
 278             cur_packet.record.csv.fields[p]=&line[i];
 279             p++;
 280             while((line[i] != 0) && (line[i] != csvdel)) {
 281                 i++;
 282             }
 283             if(line[i] != 0){
 284                 line[i]=0;
 285                 i++;
 286             }
 287         }
 288         //                      Get rid of trailing \n and \r.
 289         while(i>0 && (line[i-1] == '\n' || line[i-1] == '\r')){
 290             i--;
 291             line[i] = '\0';
 292         }
 293         cur_packet.record.csv.numberfields=p;
 294         rts_fta_process_packet(&cur_packet);
 295     }
 296
 297
 298
 299
 300
 301     static gs_retval_t csv_process_file()
 302     {
 303         unsigned cnt=0;
 304         static unsigned totalcnt=0;
 305         for(cnt=0;cnt<50000;cnt++) {
 306             if (gshub!=0) {
 307                 gs_retval_t retval;
 308                 retval=csv_read_socket();
 309                 if (retval==-1) return 0; // got a timeout so service message queue
 310                 if ((retval==-2) || (ftaschema_is_eof_tuple(cur_packet.record.gdat.schema,(void *)cur_packet.record.gdat.data))) {
 311                     // we signal that everything is done if we either see an EOF tuple OR the socket is closed by the peer
 312                     if (verbose)
 313                         fprintf(stderr,"Done processing waiting for things to shut down\n");
 314                     rts_fta_done();
 315                     // now just service message queue until we get killed or loose connectivity
 316                     while (0==0) {
 317                         fta_start_service(0); // service all waiting messages
 318                         usleep(1000); // sleep a millisecond
 319                     }
 320                 }
 321             } else {
 322                 csv_read_tuple();
 323             }
 324         }
 325         totalcnt=totalcnt+cnt;
 326         if (verbose) {
 327             fprintf(stderr,"Processesd %u tuple\n",totalcnt);
 328         }
 329         return 0;
 330     }
 331
 332     gs_retval_t main_csv(gs_int32_t devicenum, gs_sp_t device, gs_int32_t mapcnt, gs_sp_t map[]) {
 333         gs_uint32_t cont;
 334         endpoint mygshub;
 335
 336         csv_replay_init(device);
 337
 338         /* initalize host_lib */
 339         if (verbose) {
 340             fprintf(stderr,"Init LFTAs for %s\n",device);
 341         }
 342
 343         if (hostlib_init(LFTA,0,devicenum,mapcnt,map)<0) {
 344             fprintf(stderr,"%s::error:could not initiate host lib for clearinghouse\n",
 345                     device);
 346             exit(7);
 347         }
 348
 349         fta_init(device); /*xxx probably should get error code back put Ted doesn't give me one*/
 350
 351         cont=startupdelay+time(0);
 352
 353         if (verbose) { fprintf(stderr,"Start startup delay"); }
 354
 355         while (cont>time(NULL)) {
 356             if (fta_start_service(0)<0) {
 357                 fprintf(stderr,"%s::error:in processing the msg queue\n",
 358                         device);
 359                 exit(9);
 360             }
 361             usleep(1000); /* sleep for one millisecond */
 362         }
 363
 364         if (verbose) { fprintf(stderr,"... Done\n"); }
 365
 366         // open the connection to the data source
 367         if (gshub!=0) { init_socket();}
 368
 369         // wait to process till we get the signal from GSHUB
 370         if (get_hub(&mygshub)!=0) {
 371             print_error("ERROR:could not find gshub for data source");
 372             exit(0);
 373         }
 374         while(get_startprocessing(mygshub,get_instance_name(),0)!=0) {
 375             usleep(100);
 376             if (fta_start_service(0)<0) {
 377                 fprintf(stderr,"%s::error:in processing the msg queue\n",
 378                         device);
 379                 exit(9);
 380             }
 381         }
 382
 383         /* now we enter an endless loop to process data */
 384         if (verbose) {
 385             fprintf(stderr,"Start processing %s\n",device);
 386         }
 387
 388         while (1==1) {
 389             if (csv_process_file()<0) {
 390                 fprintf(stderr,"%s::error:in processing packets\n",
 391                         device);
 392                 exit(8);
 393             }
 394             /* process all messages on the message queue*/
 395             if (fta_start_service(0)<0) {
 396                 fprintf(stderr,"%s::error:in processing the msg queue\n",
 397                         device);
 398                 exit(9);
 399             }
 400         }
 401         return 0;
 402     }
 403
 404