src/tools/gssource.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 <stdlib.h>
  16 #include <stdio.h>
  17 #include <unistd.h>
  18 #include <signal.h>
  19 #include <time.h>
  20 #include <string.h>
  21 #include <sys/time.h>
  22 #include <sys/stat.h>
  23 #include <sys/types.h>
  24 #include <sys/socket.h>
  25 #include <netinet/in.h>
  26
  27 #include "gsconfig.h"
  28 #include "gstypes.h"
  29 #include "gshub.h"
  30
  31
  32
  33
  34 gs_sp_t me = 0;
  35 gs_sp_t schematext = 0;
  36 gs_int32_t schematextlen = 0;
  37 gs_sp_t schematmp = 0;
  38 gs_int32_t verbose=0;
  39 gs_uint32_t tcpport=0;
  40 int listensockfd=0;
  41 int fd=0;
  42 endpoint hub;
  43 endpoint ds;
  44 gs_sp_t source_name;
  45
  46 static void gs_write(gs_sp_t buffer, gs_uint32_t len)
  47 {
  48     if (send(fd,buffer,len,0) != len) {
  49         fprintf(stderr,"could not write on stream socket");
  50         exit(0);
  51     }
  52 }
  53
  54 static void wait_for_feed() {
  55     struct sockaddr_in serv_addr,cli_addr;
  56     struct sockaddr_in sin;
  57     socklen_t clilen;
  58     socklen_t sin_sz;
  59     if (listensockfd==0) {
  60                 gs_int32_t on = 1;
  61
  62         if (verbose) {
  63             fprintf(stderr,"Create listen socket for port %u\n",tcpport);
  64         }
  65                 listensockfd=socket(AF_INET, SOCK_STREAM, 0);
  66         if (listensockfd < 0) {
  67                         fprintf(stderr,"Error:Could not create socket for tcp data stream");
  68                         exit(1);
  69                 }
  70                 bzero((char *) &serv_addr, sizeof(serv_addr));
  71                 serv_addr.sin_family = AF_INET;
  72                 serv_addr.sin_addr.s_addr = INADDR_ANY;
  73                 serv_addr.sin_port = htons(tcpport);
  74 #ifndef __linux__
  75         /* make sure we can reuse the common port rapidly */
  76         if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEPORT,
  77                        (gs_sp_t )&on, sizeof(on)) != 0) {
  78             fprintf(stderr,"Error::could not set socket option\n");
  79             exit(1);
  80         }
  81 #endif
  82         if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEADDR,
  83                        (gs_sp_t )&on, sizeof(on)) != 0) {
  84             fprintf(stderr,"Error::could not set socket option\n");
  85             exit(1);
  86                 }
  87
  88                 if (bind(listensockfd, (struct sockaddr *) &serv_addr,
  89                  sizeof(serv_addr)) < 0) {
  90                         fprintf(stderr,"Error:Could not bind socket for tcp data stream");
  91             exit(1);
  92         }
  93         }
  94
  95     if (verbose) {
  96         fprintf(stderr,"Socket created waiting for data producer\n");
  97     }
  98     if (listen(listensockfd,5)< 0) {
  99         fprintf(stderr,"Error::could not listen to socket for port %u \n",ntohs(serv_addr.sin_port));
 100         close(listensockfd);
 101         exit(1);
 102     }
 103     sin_sz=sizeof(sin);
 104     if (getsockname(listensockfd, (struct sockaddr *) &sin, &sin_sz) < 0) {
 105         fprintf(stderr,"Error::could not get local port number of listen socket\n");
 106         exit(1);
 107     }
 108     ds.ip=htonl(127<<24|1);
 109     ds.port=sin.sin_port;
 110     if (set_streamsource(hub,source_name,ds)!=0) {
 111         fprintf(stderr,"Error::could not set source in GSHUB for %s source name\n",source_name);
 112         exit(1);
 113     }
 114
 115         do {
 116                 clilen = sizeof(cli_addr);
 117                 fd=accept(listensockfd, (struct sockaddr *) &cli_addr, &clilen);
 118                 if (fd<0) {
 119             fprintf(stderr,"Error:Could not accept connection on tcp socket\n");
 120                 }
 121         } while (fd==0);
 122     if (verbose) {
 123         fprintf(stderr,"Sink found ready to rock!\n");
 124     }
 125
 126 }
 127
 128
 129 static void do_file(gs_sp_t filename, gs_int32_t fnlen);
 130
 131 int main(int argc, char** argv) {
 132     gs_int32_t x;
 133     gs_int32_t s=0;
 134     gs_int32_t ch;
 135     gs_int32_t endless=0; // repeats files forever
 136     gs_uint32_t tip1,tip2,tip3,tip4;
 137     while ((ch = getopt(argc, argv, "hxep:")) != -1) {
 138         switch(ch) {
 139             case 'h':
 140                 fprintf(stderr,"%s::usage: %s -x -v -e -p <portnumber> <IP>:<port> <source_name> <gdatfiles...>\n",argv[0],argv[0]);
 141                 exit(0);
 142                 break;
 143             case 'p':
 144                 tcpport=atoi(optarg);;
 145                 break;
 146             case 'v':
 147                 verbose=1;
 148                 break;
 149             case 'x':
 150                 verbose=2;
 151                 break;
 152             case 'e':
 153                 endless=1;
 154                 break;
 155             default:
 156                 break;
 157         }
 158     }
 159     s+=optind;
 160     if (s+2>argc) {
 161         fprintf(stderr,"Could not find hub and stream source name on command line\n");
 162         fprintf(stderr,"%s::usage: %s -x -v -e -p <portnumber> <IP>:<port> <source_name> <gdatfiles...>\n",argv[0],argv[0]);
 163         exit(1);
 164     }
 165     if (sscanf(argv[s],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(hub.port))!= 5 ) {
 166         fprintf(stderr,"Could not parse hub endpoint\n");
 167         fprintf(stderr,"%s::usage: %s -x -v -e -p <portnumber> <IP>:<port> <source_name> <gdatfiles...>\n",argv[0],argv[0]);
 168         exit(1);
 169     }
 170     hub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
 171     hub.port=htons(hub.port);
 172     source_name=strdup(argv[s+1]);
 173     s+=2;
 174     wait_for_feed();
 175     do {
 176         for(x=s;x<argc;x++) {
 177             if (verbose) {
 178                 fprintf(stderr,"%s\n",argv[x]);
 179             }
 180             do_file(argv[x], strlen(argv[x]));
 181         }
 182     } while (endless !=0); // will run forever if endless option is set
 183     close(fd); // make sure we wait till buffers are empty
 184     return 0;
 185 }
 186
 187
 188 /*
 189  * do_file: dump the file out
 190  */
 191
 192 static void do_file(gs_sp_t filename, gs_int32_t fnlen) {
 193     gs_int32_t pipe, parserversion, schemalen;
 194     FILE *input;
 195     gs_int8_t cmd2[4096 + 128];
 196     static gs_int8_t *dbuf;
 197     size_t sz;
 198
 199     if (fnlen > 3 && filename[fnlen - 3] == '.' &&
 200         filename[fnlen - 2] == 'g' &&
 201         filename[fnlen - 1] == 'z') {
 202         pipe = 1;
 203         snprintf(cmd2, sizeof(cmd2), "gzcat %s", filename);
 204         input = popen(cmd2, "r");
 205     } else {
 206         if (fnlen > 3 && filename[fnlen - 3] == 'b' &&
 207             filename[fnlen - 2] == 'z' &&
 208             filename[fnlen - 1] == '2') {
 209             pipe = 1;
 210             snprintf(cmd2, sizeof(cmd2), "bzcat %s", filename);
 211             input = popen(cmd2, "r");
 212         } else {
 213             pipe=0;
 214             input = fopen(filename, "r");
 215         }
 216     }
 217
 218     if (!input) {
 219         perror("stream open");
 220         fprintf(stderr, "%s: cannot open %s\n", me, filename);
 221         return;
 222     }
 223
 224     if (fscanf(input, "GDAT\nVERSION:%u\nSCHEMALENGTH:%u\n",
 225                &parserversion,&schemalen) != 2) {
 226         fprintf(stderr,"%s: cannot parse GDAT file header in '%s'\n",
 227                 me, filename);
 228         exit(1);
 229     }
 230
 231     /* first time ? */
 232     if (schematext == 0) {
 233         gs_uint8_t buf[1024];
 234         schematextlen = schemalen;
 235         schematext = malloc(schemalen);
 236         dbuf = malloc(CATBLOCKSZ);
 237         if (!schematext  || !dbuf) {
 238             fprintf(stderr,"%s: malloc error reading GDAT file header in '%s'\n",
 239                     me, filename);
 240             exit(1);
 241         }
 242         if (fread(schematext, schemalen, 1, input) != 1) {
 243             fprintf(stderr,"%s: cannot parse-read GDAT file header in '%s'\n",
 244                     me, filename);
 245             exit(1);
 246         }
 247         sprintf((char *)buf,"GDAT\nVERSION:%u\nSCHEMALENGTH:%u\n", parserversion, schemalen);
 248         gs_write((gs_sp_t)buf,strlen((const char*)buf));
 249         gs_write(schematext, schemalen);
 250     } else {
 251         schematmp = malloc(schemalen);
 252         if (!schematmp ) {
 253             fprintf(stderr,"%s: malloc error reading GDAT file header in '%s'\n",
 254                     me, filename);
 255             exit(1);
 256         }
 257         if (fread(schematmp, schemalen, 1, input) != 1) {
 258             fprintf(stderr,"%s: cannot parse-read GDAT file header in '%s'\n",
 259                     me, filename);
 260             exit(1);
 261         }
 262         free(schematmp);
 263         //   if (memcmp(schematext, schematmp, schematextlen)) {
 264         //     fprintf(stderr,"%s: GDAT schema mis-match in file '%s'\n",
 265         //             me, filename);
 266         //     exit(1);
 267         //   }
 268     }
 269
 270     while ((sz = fread(dbuf, 1, CATBLOCKSZ, input)) > 0) {
 271         gs_write(dbuf,sz);
 272     }
 273
 274     if (pipe) {
 275         pclose(input);
 276     } else {
 277         fclose(input);
 278     }
 279
 280 }