-/* ------------------------------------------------
- Copyright 2014 AT&T Intellectual Property
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ------------------------------------------- */
-#include <app.h>
-#include <stdlib.h>
-#include <string.h>
-#include <stdio.h>
-#include <unistd.h>
-#include <signal.h>
-#include <time.h>
-#include <string.h>
-#include <sys/time.h>
-#include <sys/stat.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-
-#include "gsconfig.h"
-#include "gstypes.h"
-#include <schemaparser.h>
-#include <gshub.h>
-
-#define MAXSTRLEN 256
-#define MAXNUMFIELDS 64
-
-//#define GSBZIP2
-
-#ifdef GSBZIP2
-#include "bzlib.h"
-#endif
-
-#ifdef ZLIB
-#include "zlib.h"
-#endif
-
-gs_sp_t me; /* saved copy argv[0] */
-
-struct FTA_state {
- FTAID fta_id;
- gs_schemahandle_t schema;
- gs_sp_t asciischema;
- gs_int32_t numfields;
- gs_int32_t timefieldoffset;
-};
-
-struct gsasciiprint_state {
- // configuration state
- gs_uint32_t bufsz;
- gs_int32_t stream;
- gs_int32_t flush;
- gs_int32_t parserversion;
- gs_int32_t compressed;
- gs_int32_t verbose;
- gs_int32_t notemp;
- gs_sp_t timefield;
- gs_int32_t interval;
- gs_int32_t quitcnt;
- gs_int32_t quittime;
- gs_int32_t numfields;
- gs_sp_t extension;
- gs_sp_t query;
- gs_int32_t remote_print;
-
-
- // runtime state
- struct FTA_state fs;
-};
-
-
-void hand(int iv) {
- ftaapp_exit();
- gslog(LOG_NOTICE, "exiting via signal handler %d...\n", iv);
- exit(0);
-}
-
-void timeouthand(int iv) {
- ftaapp_exit();
- // if (s->verbose!=0) fprintf(stderr, "exiting because of timeout...\n");
- exit(0);
-}
-
-
-static void print_usage_exit(gs_sp_t reason) {
- fprintf(stderr,
- "%s::error: %s\n"
- "%s::usage: %s -r <int> -v -f -s -z -c <int> -q <int> -b <field> -t <int> -e <string> <gshub-hostname>:<gshub-port> <gsinstance_name> <query_name> <parameters>\n"
- "\t-v makes output verbose\n"
- "\t-f flushes each tuple individually\n"
- "\t-h <int> print within the hfta int identifies how many output streams based on the SubInterface field\n"
- "\t-r sets the ringbuffer size default is 8MB\n"
- "\t-s uses %s in streaming mode cannot be used with -v -z -b -t -e\n"
- "\t-c <int> indicates that %s should terminate after <int> tuples.\n"
- "\t-q <int> indicates that %s should terminate after <int> seconds.\n"
- "\t-z the output is compressed with gzip \n"
- "\t-b <field> identifies the field which is increasing and is used to\n"
- "\t\tbin output files. Field has to be of type uint.\n"
- "\t\tThe default is the UNIX system time\n"
- "\t-t <int> specifies the number the <field> specified with -b\n"
- "\t\thas to increase before a new output file is used\n"
- "\t\tthe default is 60 for 1 minute if the default UNIX time is used\n"
- "\t-e <string> identifies the file string extension used for the \n"
- "\t\toutput file. The output file always starts with the current value\n"
- "\t\tof the b field (or UNIX time) the default extension is .txt \n"
- "\t\twithout the -z option and .txt.gz with the -z option\n"
- "\t<query_name> specifies the query which will be\n"
- "\t\t instanciated.\n"
- "\t<parameters> sets the parameters for the query\n"
- , me, reason, me, me, me,me,me);
- exit(1);
-}
-
-
-
-
-static void init(gs_int32_t argc, gs_sp_t argv[], struct gsasciiprint_state *s) {
- void *pblk;
- gs_int32_t x, y, schema, pblklen, lcv;
- gs_int32_t n_actual_param;
- gs_int32_t n_expected_param;
- gs_sp_t c;
- gs_int8_t name[1024];
-
- sprintf(name,"gsgdatprint:%s",argv[0]);
-
- gsopenlog(name);
-
-
- if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing gscp\n");
- if (ftaapp_init(s->bufsz)!=0) {
- gslog(LOG_EMERG,"%s::error:could not initialize gscp\n", me);
- exit(1);
- }
-
- signal(SIGTERM, hand);
- signal(SIGINT, hand);
- signal(SIGPIPE, hand);
-
- if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing FTAs\n");
-
- schema = ftaapp_get_fta_schema_by_name(argv[0]);
- if (schema < 0) {
- gslog(LOG_EMERG,"%s::error:can't get fta '%s' schema\n", me ,argv[0]);
- exit(1);
- }
- n_expected_param = ftaschema_parameter_len(schema);
- if (n_expected_param == 0) {
- pblk = 0;
- pblklen = 0;
- if (s->verbose) gslog(LOG_DEBUG,"[query does not have any params]\n");
- } else {
- n_actual_param = argc-1;
- if(n_actual_param < n_expected_param){
- gslog(LOG_EMERG,"Error, %d query parameters expected, %d provided.\n",n_expected_param, n_actual_param);
- exit(1);
- }
- /* parse the params */
- for (lcv = 1 ; lcv < argc ; lcv++) {
- char *k, *e;
- int rv;
- k = argv[lcv];
- e = k;
- while (*e && *e != '=') e++;
- if (*e == 0) {
- gslog(LOG_EMERG,"param parse error '%s' (fmt 'key=val')\n",
- argv[lcv]);
- exit(1);
- }
- *e = 0;
- rv = ftaschema_setparam_by_name(schema, k, e+1, strlen(e+1));
- *e = '=';
- if (rv < 0) {
- gslog(LOG_EMERG,"param setparam error '%s' (fmt 'key=val')\n",
- argv[lcv]);
- exit(1);
- }
- }
- if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {
- gslog(LOG_EMERG, "ftaschema_create_param_block failed!\n");
- exit(1);
- }
- }
- ftaschema_free(schema); /* XXXCDC */
-
-
- if (s->remote_print>=0) {
- s->fs.fta_id=ftaapp_add_fta_print(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk,"./data",s->extension,s->timefield,"SubInterface",s->interval,s->remote_print);
- if (s->fs.fta_id.streamid==0){
- gslog(LOG_EMERG,"%s::error:could not initialize fta_print %s\n",
- me,s->query);
- exit(1);
- }
- again:
- // wait forever
- sleep(60);
- goto again;
- }
-
-
- s->fs.fta_id=ftaapp_add_fta(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk);
- if (s->fs.fta_id.streamid==0){
- gslog(LOG_EMERG,"%s::error:could not initialize fta %s\n",
- me,s->query);
- exit(1);
- }
- /* XXXCDC: pblk is malloc'd, should we free it? */
-
- if ((c=ftaapp_get_fta_ascii_schema_by_name(s->query))==0){
- gslog(LOG_EMERG,"%s::error:could not get ascii schema for %s\n",
- me,s->query);
- exit(1);
- }
-
- //ftaapp_get_fta_ascii_schema_by_name uses static buffer so make a copy
- s->fs.asciischema=strdup(c);
-
-
- // Set parser version here
- s->parserversion=get_schemaparser_version();
-
- if ((s->fs.schema=ftaapp_get_fta_schema(s->fs.fta_id))<0) {
- gslog(LOG_EMERG,"%s::error:could not get schema for query\n",
- me,s->query);
- exit(1);
- }
-
- // Use all available fields
- if ((s->fs.numfields=ftaschema_tuple_len(s->fs.schema))<0) {
- gslog(LOG_EMERG,"%s::error:could not get number of fields for query %s\n",
- me,s->query);
- exit(1);
- }
- if (s->timefield!=0) {
- if ((s->fs.timefieldoffset=ftaschema_get_field_offset_by_name(
- s->fs.schema,s->timefield))<0) {
- gslog(LOG_EMERG,"%s::error:could not get "
- "offset for timefield %s in query %s\n",
- me,s->timefield,s->query);
- exit(1);
- }
-
- if (ftaschema_get_field_type_by_name(
- s->fs.schema,s->timefield)!=UINT_TYPE) {
- gslog(LOG_EMERG,"%s::error: illegal type for timefield "
- "%s in query %s UINT expected\n",
- me,s->timefield,s->query);
- exit(1);
- }
- }
-}
-
-
-static void process_data(struct gsasciiprint_state *s)
-{
- gs_int32_t x;
- gs_int32_t y;
- gs_int32_t z;
- gs_int32_t tb;
- gs_uint32_t nsz;
- FTAID rfta_id;
- gs_uint32_t rsize;
- gs_int8_t rbuf[2*MAXTUPLESZ];
- gs_int32_t problem;
-
- gs_int32_t ctb=0;
- gs_int8_t fname[1024];
- gs_int8_t tmpname[1024];
- gs_int8_t command[1024];
- gs_int8_t topb[1024];
- gs_int32_t rcnt=0;
- gs_retval_t code;
-#ifdef GSBZIP2
- int bzerror;
- BZFILE * b;
- int abandon=0;
- unsigned int bytes_in;
- unsigned int bytes_out;
-#endif
-#ifdef ZLIB
- gzFile of=0;
-#else
- FILE * of=0;
-#endif
-
-
- tmpname[0]=0;
-
- if (s->verbose!=0) gslog(LOG_INFO,"Getting Data for %s",s->query);
-
- sprintf(&topb[0],"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n",
- s->parserversion,strlen(s->fs.asciischema)+1);
- if (s->stream!=0) {
- of=stdout;
- // need to get ASCII version of schema
- fwrite(&topb[0],strlen(topb),1,of);
- fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);
- }
-
- if (s->quittime !=0 ) {
- signal(SIGALRM,timeouthand);
- alarm(s->quittime);
- }
-
-
- while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {
- rcnt++;
- if ((s->notemp==1) && (code==2)) continue;
- if (((s->quitcnt>0) && (rcnt>s->quitcnt))||ftaschema_is_eof_tuple(s->fs.schema, rbuf)) {
- if (s->verbose!=0)
- gslog(LOG_EMERG, "exiting reached tuple limit or all data has been proccessed\n");
- if (of!=0) {
- fclose(of);
-
- if (s->compressed) {
- system(command);
- }
- rename(tmpname,fname);
- }
- exit(0);
- }
- if (((code==0) || (code==2))&&(s->fs.fta_id.streamid==rfta_id.streamid)) {
- if (s->stream==0) {
- if (s->timefield!=0) {
- tb=fta_unpack_uint(rbuf,
- rsize,
- s->fs.timefieldoffset,
- &problem);
- } else {
- tb=time(0);
- }
- if ((ctb+s->interval)<=tb) {
- gsstats();
- if (of!=0) {
-#ifdef GSBZIP2
- BZ2_bzWriteClose(&bzerror, b, abandon, &bytes_in, &bytes_out );
- if (bzerror!=BZ_OK) {
- gslog(LOG_EMERG,"Could not bz close file .. EXITING\n");
- exit(0);
- }
-#endif
-#ifdef ZLIB
- gzclose(of);
-#else
- fclose(of);
-#endif
-#ifndef GSBZIP2
- if (s->compressed) {
- system(command);
- }
-#endif
- rename(tmpname,fname);
- }
- while((ctb+s->interval)<=tb) {
- if (ctb==0) {
- ctb=(tb/s->interval)*s->interval;
- } else {
- ctb=ctb+s->interval;
- }
- }
-#ifdef ZLIB
- sprintf(tmpname,"%u%s.gz.tmp",ctb,s->extension);
- sprintf(fname,"%u%s.gz",ctb,s->extension);
-#else
- sprintf(tmpname,"%u%s.tmp",ctb,s->extension);
- sprintf(fname,"%u%s",ctb,s->extension);
-#endif
-#ifndef GSBZIP2
- if (s->compressed) {
- sprintf(command,"gzip -S .tmpgz %s ; mv %s.tmpgz %s",tmpname,tmpname,tmpname);
- }
-#endif
-#ifdef ZLIB
- if ((of=gzopen(tmpname,"wb"))==0) {
- gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",
- tmpname);
- exit(0);
- }
-#else
- if ((of=fopen(tmpname,"w"))==0) {
- gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",
- tmpname);
- exit(0);
- }
-#endif
-#ifdef GSBZIP2
- if (s->compressed) {
- b=BZ2_bzWriteOpen(&bzerror,of,5,0,30);
- if (bzerror!=BZ_OK) {
- gslog(LOG_EMERG,"Could not bz open file \"%s\".. EXITING\n",
- tmpname);
- exit(0);
- }
- }
-#endif
-
- if (s->compressed) {
-#ifdef GSBZIP2
- BZ2_bzWrite ( &bzerror, b, &topb[0],strlen(topb));
- BZ2_bzWrite ( &bzerror, b, s->fs.asciischema,strlen(s->fs.asciischema)+1);
-#endif
- } else {
- // need to get ASCII version of schema
-#ifdef ZLIB
- gzwrite(of,&topb[0],strlen(topb));
- gzwrite(of,s->fs.asciischema,strlen(s->fs.asciischema)+1);
-#else
- fwrite(&topb[0],strlen(topb),1,of);
- fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);
-#endif
- }
- }
- }
- if (code==0) {
- nsz=htonl(rsize);
- if (s->compressed) {
-#ifdef GSBZIP2
- BZ2_bzWrite ( &bzerror, b,&nsz,sizeof(gs_uint32_t) );
- BZ2_bzWrite ( &bzerror, b,rbuf,rsize);
-#endif
- } else {
-#ifdef ZLIB
- gzwrite(of,&nsz,sizeof(gs_uint32_t));
- gzwrite(of,rbuf,rsize);
- }
-#else
- if (fwrite(&nsz,sizeof(gs_uint32_t),1,of)!=1) {
- ftaapp_exit();
- if (s->verbose!=0)
- gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",
- tmpname);
- exit(0);
- }
-
- if (fwrite(rbuf,rsize,1,of)!=1) {
- ftaapp_exit();
- if (s->verbose!=0)
- gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",
- tmpname);
- exit(0);
- }
- if ((s->stream!=0) || (s->flush!=0)) {
- fflush(of);
- }
- }
-#endif
- }
- }
-}
-}
-
-
-
-int main(int argc, char** argv) {
-
- struct gsasciiprint_state s;
- gs_retval_t ch;
- endpoint gshub;
- endpoint dummyep;
- gs_uint32_t tip1,tip2,tip3,tip4;
- gs_sp_t instance_name;
-
- me = argv[0];
-
- /* initialize host library and the sgroup */
-
- if (argc<=1) {
- print_usage_exit("Not enough arguments");
- }
-
- /* parse args */
- bzero(&s, sizeof(s));
- s.interval = 60; /* default */
- s.quittime=0;
- s.quitcnt=-1;
- s.remote_print=-1;
- s.bufsz=8*1024*1024;
-
- while ((ch = getopt(argc, argv, "nr:b:e:t:c:q:sfvzmh:")) != -1) {
- switch(ch) {
- case 'r':
- s.bufsz=atoi(optarg);
- break;
- case 'c':
- s.quitcnt = atoi(optarg);
- break;
- case 'q':
- s.quittime = atoi(optarg);
- break;
- case 'b':
- s.timefield = optarg;
- break;
- case 'e':
- s.extension = optarg;
- break;
- case 'n':
- s.notemp=1;
- break;
- case 't':
- s.interval = atoi(optarg);
- if (s.interval < 1) {
- goto usage;
- }
- break;
- case 's':
- s.stream++;
- break;
- case 'f':
- s.flush++;
- break;
- case 'v':
- s.verbose++;
- break;
- case 'z':
- s.compressed++;
- break;
- case 'h':
- s.remote_print=atoi(optarg);
- break;
- default:
- usage:
- print_usage_exit("invalid args");
- }
- }
-
- if ((s.stream!=0) & (s.compressed!=0 | s.verbose!=0 |
- s.timefield!=0 | s.extension!=0)) {
- print_usage_exit("illegal argument combination with -s");
- }
-
- if (!s.extension) {
- s.extension = (s.compressed) ? ".txt.gz" : ".txt";
- }
-
- argc -= optind;
- argv += optind;
- if (argc<3) print_usage_exit("must specify hub info and query");
-
- if (sscanf(argv[0],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(gshub.port))!= 5 ) {
- gslog(LOG_EMERG,"HUB IP NOT DEFINED");
- exit(1);
- }
- gshub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
- gshub.port=htons(gshub.port);
- instance_name=strdup(argv[1]);
- if (set_hub(gshub)!=0) {
- gslog(LOG_EMERG,"Could not set hub");
- exit(1);
- }
- if (set_instance_name(instance_name)!=0) {
- gslog(LOG_EMERG,"Could not set instance name");
- exit(1);
- }
-
- if (get_initinstance(gshub,instance_name,&dummyep,1)!=0) {
- gslog(LOG_EMERG,"Did not receive signal that GS is initiated\n");
- }
-
-
- argc -=2;
- argv +=2;
-
- s.query = argv[0];
-
- init(argc, argv, &s);
-
- process_data(&s);
-
- gslog(LOG_EMERG,"%s::internal error reached unexpected end", me);
-}
-
+/* ------------------------------------------------\r
+ Copyright 2014 AT&T Intellectual Property\r
+ Licensed under the Apache License, Version 2.0 (the "License");\r
+ you may not use this file except in compliance with the License.\r
+ You may obtain a copy of the License at\r
+ \r
+ http://www.apache.org/licenses/LICENSE-2.0\r
+ \r
+ Unless required by applicable law or agreed to in writing, software\r
+ distributed under the License is distributed on an "AS IS" BASIS,\r
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ See the License for the specific language governing permissions and\r
+ limitations under the License.\r
+ ------------------------------------------- */\r
+#include <app.h>\r
+#include <stdlib.h>\r
+#include <string.h>\r
+#include <stdio.h>\r
+#include <unistd.h>\r
+#include <signal.h>\r
+#include <time.h>\r
+#include <string.h>\r
+#include <sys/time.h>\r
+#include <sys/stat.h>\r
+#include <sys/types.h>\r
+#include <sys/socket.h>\r
+#include <netinet/in.h>\r
+\r
+#include "gsconfig.h"\r
+#include "gstypes.h"\r
+#include <schemaparser.h>\r
+#include <gshub.h>\r
+\r
+#define MAXSTRLEN 256\r
+#define MAXNUMFIELDS 64\r
+\r
+//#define GSBZIP2\r
+\r
+#ifdef GSBZIP2\r
+#include "bzlib.h"\r
+#endif\r
+\r
+#ifdef ZLIB\r
+#include "zlib.h"\r
+#endif\r
+\r
+gs_sp_t me; /* saved copy argv[0] */\r
+\r
+struct FTA_state {\r
+ FTAID fta_id;\r
+ gs_schemahandle_t schema;\r
+ gs_sp_t asciischema;\r
+ gs_int32_t numfields;\r
+ gs_int32_t timefieldoffset;\r
+};\r
+\r
+struct gsasciiprint_state {\r
+ // configuration state\r
+ gs_uint32_t bufsz;\r
+ gs_int32_t stream;\r
+ gs_int32_t flush;\r
+ gs_int32_t parserversion;\r
+ gs_int32_t compressed;\r
+ gs_int32_t verbose;\r
+ gs_int32_t notemp;\r
+ gs_sp_t timefield;\r
+ gs_int32_t interval;\r
+ gs_int32_t quitcnt;\r
+ gs_int32_t quittime;\r
+ gs_int32_t numfields;\r
+ gs_sp_t extension;\r
+ gs_sp_t query;\r
+ gs_int32_t remote_print;\r
+ \r
+ \r
+ // runtime state\r
+ struct FTA_state fs;\r
+};\r
+\r
+\r
+void hand(int iv) {\r
+ ftaapp_exit();\r
+ gslog(LOG_NOTICE, "exiting via signal handler %d...\n", iv);\r
+ exit(0);\r
+}\r
+\r
+void timeouthand(int iv) {\r
+ ftaapp_exit();\r
+ // if (s->verbose!=0) fprintf(stderr, "exiting because of timeout...\n");\r
+ exit(0);\r
+}\r
+\r
+\r
+static void print_usage_exit(gs_sp_t reason) {\r
+ fprintf(stderr,\r
+ "%s::error: %s\n"\r
+ "%s::usage: %s -r <int> -v -f -s -z -c <int> -q <int> -b <field> -t <int> -e <string> <gshub-hostname>:<gshub-port> <gsinstance_name> <query_name> <parameters>\n"\r
+ "\t-v makes output verbose\n"\r
+ "\t-f flushes each tuple individually\n"\r
+ "\t-h <int> print within the hfta int identifies how many output streams based on the SubInterface field\n"\r
+ "\t-r sets the ringbuffer size default is 8MB\n"\r
+ "\t-s uses %s in streaming mode cannot be used with -v -z -b -t -e\n"\r
+ "\t-c <int> indicates that %s should terminate after <int> tuples.\n"\r
+ "\t-q <int> indicates that %s should terminate after <int> seconds.\n"\r
+ "\t-z the output is compressed with gzip \n"\r
+ "\t-b <field> identifies the field which is increasing and is used to\n"\r
+ "\t\tbin output files. Field has to be of type uint.\n"\r
+ "\t\tThe default is the UNIX system time\n"\r
+ "\t-t <int> specifies the number the <field> specified with -b\n"\r
+ "\t\thas to increase before a new output file is used\n"\r
+ "\t\tthe default is 60 for 1 minute if the default UNIX time is used\n"\r
+ "\t-e <string> identifies the file string extension used for the \n"\r
+ "\t\toutput file. The output file always starts with the current value\n"\r
+ "\t\tof the b field (or UNIX time) the default extension is .txt \n"\r
+ "\t\twithout the -z option and .txt.gz with the -z option\n"\r
+ "\t<query_name> specifies the query which will be\n"\r
+ "\t\t instanciated.\n"\r
+ "\t<parameters> sets the parameters for the query\n"\r
+ , me, reason, me, me, me,me,me);\r
+ exit(1);\r
+}\r
+\r
+\r
+\r
+\r
+static void init(gs_int32_t argc, gs_sp_t argv[], struct gsasciiprint_state *s) {\r
+ void *pblk;\r
+ gs_int32_t x, y, schema, pblklen, lcv;\r
+ gs_int32_t n_actual_param;\r
+ gs_int32_t n_expected_param;\r
+ gs_sp_t c;\r
+ gs_int8_t name[1024];\r
+ \r
+ sprintf(name,"gsgdatprint:%s",argv[0]);\r
+ \r
+ gsopenlog(name);\r
+ \r
+ \r
+ if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing gscp\n");\r
+ if (ftaapp_init(s->bufsz)!=0) {\r
+ gslog(LOG_EMERG,"%s::error:could not initialize gscp\n", me);\r
+ exit(1);\r
+ }\r
+ \r
+ signal(SIGTERM, hand);\r
+ signal(SIGINT, hand);\r
+ signal(SIGPIPE, hand);\r
+ \r
+ if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing FTAs\n");\r
+ \r
+ schema = ftaapp_get_fta_schema_by_name(argv[0]);\r
+ if (schema < 0) {\r
+ gslog(LOG_EMERG,"%s::error:can't get fta '%s' schema\n", me ,argv[0]);\r
+ exit(1);\r
+ }\r
+ n_expected_param = ftaschema_parameter_len(schema);\r
+ if (n_expected_param == 0) {\r
+ pblk = 0;\r
+ pblklen = 0;\r
+ if (s->verbose) gslog(LOG_DEBUG,"[query does not have any params]\n");\r
+ } else {\r
+ n_actual_param = argc-1;\r
+ if(n_actual_param < n_expected_param){\r
+ gslog(LOG_EMERG,"Error, %d query parameters expected, %d provided.\n",n_expected_param, n_actual_param);\r
+ exit(1);\r
+ }\r
+ /* parse the params */\r
+ for (lcv = 1 ; lcv < argc ; lcv++) {\r
+ char *k, *e;\r
+ int rv;\r
+ k = argv[lcv];\r
+ e = k;\r
+ while (*e && *e != '=') e++;\r
+ if (*e == 0) {\r
+ gslog(LOG_EMERG,"param parse error '%s' (fmt 'key=val')\n",\r
+ argv[lcv]);\r
+ exit(1);\r
+ }\r
+ *e = 0;\r
+ rv = ftaschema_setparam_by_name(schema, k, e+1, strlen(e+1));\r
+ *e = '=';\r
+ if (rv < 0) {\r
+ gslog(LOG_EMERG,"param setparam error '%s' (fmt 'key=val')\n",\r
+ argv[lcv]);\r
+ exit(1);\r
+ }\r
+ }\r
+ if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {\r
+ gslog(LOG_EMERG, "ftaschema_create_param_block failed!\n");\r
+ exit(1);\r
+ }\r
+ }\r
+ ftaschema_free(schema); /* XXXCDC */\r
+ \r
+ \r
+ if (s->remote_print>=0) {\r
+ s->fs.fta_id=ftaapp_add_fta_print(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk,"./data",s->extension,s->timefield,"SubInterface",s->interval,s->remote_print);\r
+ if (s->fs.fta_id.streamid==0){\r
+ gslog(LOG_EMERG,"%s::error:could not initialize fta_print %s\n",\r
+ me,s->query);\r
+ exit(1);\r
+ }\r
+ again:\r
+ // wait forever\r
+ sleep(60);\r
+ goto again;\r
+ }\r
+ \r
+ \r
+ s->fs.fta_id=ftaapp_add_fta(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk);\r
+ if (s->fs.fta_id.streamid==0){\r
+ gslog(LOG_EMERG,"%s::error:could not initialize fta %s\n",\r
+ me,s->query);\r
+ exit(1);\r
+ }\r
+ /* XXXCDC: pblk is malloc'd, should we free it? */\r
+ \r
+ if ((c=ftaapp_get_fta_ascii_schema_by_name(s->query))==0){\r
+ gslog(LOG_EMERG,"%s::error:could not get ascii schema for %s\n",\r
+ me,s->query);\r
+ exit(1);\r
+ }\r
+ \r
+ //ftaapp_get_fta_ascii_schema_by_name uses static buffer so make a copy\r
+ s->fs.asciischema=strdup(c);\r
+ \r
+ \r
+ // Set parser version here\r
+ s->parserversion=get_schemaparser_version();\r
+ \r
+ if ((s->fs.schema=ftaapp_get_fta_schema(s->fs.fta_id))<0) {\r
+ gslog(LOG_EMERG,"%s::error:could not get schema for query\n",\r
+ me,s->query);\r
+ exit(1);\r
+ }\r
+ \r
+ // Use all available fields\r
+ if ((s->fs.numfields=ftaschema_tuple_len(s->fs.schema))<0) {\r
+ gslog(LOG_EMERG,"%s::error:could not get number of fields for query %s\n",\r
+ me,s->query);\r
+ exit(1);\r
+ }\r
+ if (s->timefield!=0) {\r
+ if ((s->fs.timefieldoffset=ftaschema_get_field_offset_by_name(\r
+ s->fs.schema,s->timefield))<0) {\r
+ gslog(LOG_EMERG,"%s::error:could not get "\r
+ "offset for timefield %s in query %s\n",\r
+ me,s->timefield,s->query);\r
+ exit(1);\r
+ }\r
+ \r
+ if (ftaschema_get_field_type_by_name(\r
+ s->fs.schema,s->timefield)!=UINT_TYPE) {\r
+ gslog(LOG_EMERG,"%s::error: illegal type for timefield "\r
+ "%s in query %s UINT expected\n",\r
+ me,s->timefield,s->query);\r
+ exit(1);\r
+ }\r
+ }\r
+}\r
+\r
+\r
+static void process_data(struct gsasciiprint_state *s)\r
+{\r
+ gs_int32_t x;\r
+ gs_int32_t y;\r
+ gs_int32_t z;\r
+ gs_int32_t tb;\r
+ gs_uint32_t nsz;\r
+ FTAID rfta_id;\r
+ gs_uint32_t rsize;\r
+ gs_int8_t rbuf[2*MAXTUPLESZ];\r
+ gs_int32_t problem;\r
+ \r
+ gs_int32_t ctb=0;\r
+ gs_int8_t fname[1024];\r
+ gs_int8_t tmpname[1024];\r
+ gs_int8_t command[1024];\r
+ gs_int8_t topb[1024];\r
+ gs_int32_t rcnt=0;\r
+ gs_retval_t code;\r
+#ifdef GSBZIP2\r
+ int bzerror;\r
+ BZFILE * b;\r
+ int abandon=0;\r
+ unsigned int bytes_in;\r
+ unsigned int bytes_out;\r
+#endif\r
+#ifdef ZLIB\r
+ gzFile of=0;\r
+#else\r
+ FILE * of=0;\r
+#endif\r
+ \r
+ \r
+ tmpname[0]=0;\r
+ \r
+ if (s->verbose!=0) gslog(LOG_INFO,"Getting Data for %s",s->query);\r
+ \r
+ sprintf(&topb[0],"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n",\r
+ s->parserversion,strlen(s->fs.asciischema)+1);\r
+ if (s->stream!=0) {\r
+ of=stdout;\r
+ // need to get ASCII version of schema\r
+ fwrite(&topb[0],strlen(topb),1,of);\r
+ fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);\r
+ }\r
+ \r
+ if (s->quittime !=0 ) {\r
+ signal(SIGALRM,timeouthand);\r
+ alarm(s->quittime);\r
+ }\r
+ \r
+ \r
+ while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {\r
+ rcnt++;\r
+ if ((s->notemp==1) && (code==2)) continue;\r
+ if (((s->quitcnt>0) && (rcnt>s->quitcnt))||ftaschema_is_eof_tuple(s->fs.schema, rbuf)) {\r
+ if (s->verbose!=0)\r
+ gslog(LOG_EMERG, "exiting reached tuple limit or all data has been proccessed\n");\r
+ if (of!=0) {\r
+ fclose(of);\r
+ \r
+ if (s->compressed) {\r
+ system(command);\r
+ }\r
+ rename(tmpname,fname);\r
+ }\r
+ exit(0);\r
+ }\r
+ if (((code==0) || (code==2))&&(s->fs.fta_id.streamid==rfta_id.streamid)) {\r
+ if (s->stream==0) {\r
+ if (s->timefield!=0) {\r
+ tb=fta_unpack_uint(rbuf,\r
+ rsize,\r
+ s->fs.timefieldoffset,\r
+ &problem);\r
+ } else {\r
+ tb=time(0);\r
+ }\r
+ if ((ctb+s->interval)<=tb) {\r
+ gsstats();\r
+ if (of!=0) {\r
+#ifdef GSBZIP2\r
+ BZ2_bzWriteClose(&bzerror, b, abandon, &bytes_in, &bytes_out );\r
+ if (bzerror!=BZ_OK) {\r
+ gslog(LOG_EMERG,"Could not bz close file .. EXITING\n");\r
+ exit(0);\r
+ }\r
+#endif\r
+#ifdef ZLIB\r
+ gzclose(of);\r
+#else\r
+ fclose(of);\r
+#endif\r
+#ifndef GSBZIP2\r
+ if (s->compressed) {\r
+ system(command);\r
+ }\r
+#endif\r
+ rename(tmpname,fname);\r
+ }\r
+ while((ctb+s->interval)<=tb) {\r
+ if (ctb==0) {\r
+ ctb=(tb/s->interval)*s->interval;\r
+ } else {\r
+ ctb=ctb+s->interval;\r
+ }\r
+ }\r
+#ifdef ZLIB\r
+ sprintf(tmpname,"%u%s.gz.tmp",ctb,s->extension);\r
+ sprintf(fname,"%u%s.gz",ctb,s->extension);\r
+#else\r
+ sprintf(tmpname,"%u%s.tmp",ctb,s->extension);\r
+ sprintf(fname,"%u%s",ctb,s->extension);\r
+#endif\r
+#ifndef GSBZIP2\r
+ if (s->compressed) {\r
+ sprintf(command,"gzip -S .tmpgz %s ; mv %s.tmpgz %s",tmpname,tmpname,tmpname);\r
+ }\r
+#endif\r
+#ifdef ZLIB\r
+ if ((of=gzopen(tmpname,"wb"))==0) {\r
+ gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",\r
+ tmpname);\r
+ exit(0);\r
+ }\r
+#else\r
+ if ((of=fopen(tmpname,"w"))==0) {\r
+ gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",\r
+ tmpname);\r
+ exit(0);\r
+ }\r
+#endif\r
+#ifdef GSBZIP2\r
+ if (s->compressed) {\r
+ b=BZ2_bzWriteOpen(&bzerror,of,5,0,30);\r
+ if (bzerror!=BZ_OK) {\r
+ gslog(LOG_EMERG,"Could not bz open file \"%s\".. EXITING\n",\r
+ tmpname);\r
+ exit(0);\r
+ }\r
+ }\r
+#endif\r
+ \r
+ if (s->compressed) {\r
+#ifdef GSBZIP2\r
+ BZ2_bzWrite ( &bzerror, b, &topb[0],strlen(topb));\r
+ BZ2_bzWrite ( &bzerror, b, s->fs.asciischema,strlen(s->fs.asciischema)+1);\r
+#endif\r
+ } else {\r
+ // need to get ASCII version of schema\r
+#ifdef ZLIB\r
+ gzwrite(of,&topb[0],strlen(topb));\r
+ gzwrite(of,s->fs.asciischema,strlen(s->fs.asciischema)+1);\r
+#else\r
+ fwrite(&topb[0],strlen(topb),1,of);\r
+ fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);\r
+#endif\r
+ }\r
+ }\r
+ }\r
+ if (code==0) {\r
+ nsz=htonl(rsize);\r
+ if (s->compressed) {\r
+#ifdef GSBZIP2\r
+ BZ2_bzWrite ( &bzerror, b,&nsz,sizeof(gs_uint32_t) );\r
+ BZ2_bzWrite ( &bzerror, b,rbuf,rsize);\r
+#endif\r
+ } else {\r
+#ifdef ZLIB\r
+ gzwrite(of,&nsz,sizeof(gs_uint32_t));\r
+ gzwrite(of,rbuf,rsize);\r
+ }\r
+#else\r
+ if (fwrite(&nsz,sizeof(gs_uint32_t),1,of)!=1) {\r
+ ftaapp_exit();\r
+ if (s->verbose!=0)\r
+ gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",\r
+ tmpname);\r
+ exit(0);\r
+ }\r
+ \r
+ if (fwrite(rbuf,rsize,1,of)!=1) {\r
+ ftaapp_exit();\r
+ if (s->verbose!=0)\r
+ gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",\r
+ tmpname);\r
+ exit(0);\r
+ }\r
+ if ((s->stream!=0) || (s->flush!=0)) {\r
+ fflush(of);\r
+ }\r
+ }\r
+#endif\r
+ }\r
+ }\r
+}\r
+}\r
+\r
+\r
+\r
+int main(int argc, char** argv) {\r
+ \r
+ struct gsasciiprint_state s;\r
+ gs_retval_t ch;\r
+ endpoint gshub;\r
+ endpoint dummyep;\r
+ gs_uint32_t tip1,tip2,tip3,tip4;\r
+ gs_sp_t instance_name;\r
+ \r
+ me = argv[0];\r
+ \r
+ /* initialize host library and the sgroup */\r
+ \r
+ if (argc<=1) {\r
+ print_usage_exit("Not enough arguments");\r
+ }\r
+ \r
+ /* parse args */\r
+ bzero(&s, sizeof(s));\r
+ s.interval = 60; /* default */\r
+ s.quittime=0;\r
+ s.quitcnt=-1;\r
+ s.remote_print=-1;\r
+ s.bufsz=8*1024*1024;\r
+ \r
+ while ((ch = getopt(argc, argv, "nr:b:e:t:c:q:sfvzmh:")) != -1) {\r
+ switch(ch) {\r
+ case 'r':\r
+ s.bufsz=atoi(optarg);\r
+ break;\r
+ case 'c':\r
+ s.quitcnt = atoi(optarg);\r
+ break;\r
+ case 'q':\r
+ s.quittime = atoi(optarg);\r
+ break;\r
+ case 'b':\r
+ s.timefield = optarg;\r
+ break;\r
+ case 'e':\r
+ s.extension = optarg;\r
+ break;\r
+ case 'n':\r
+ s.notemp=1;\r
+ break;\r
+ case 't':\r
+ s.interval = atoi(optarg);\r
+ if (s.interval < 1) {\r
+ goto usage;\r
+ }\r
+ break;\r
+ case 's':\r
+ s.stream++;\r
+ break;\r
+ case 'f':\r
+ s.flush++;\r
+ break;\r
+ case 'v':\r
+ s.verbose++;\r
+ break;\r
+ case 'z':\r
+ s.compressed++;\r
+ break;\r
+ case 'h':\r
+ s.remote_print=atoi(optarg);\r
+ break;\r
+ default:\r
+ usage:\r
+ print_usage_exit("invalid args");\r
+ }\r
+ }\r
+ \r
+ if ((s.stream!=0) & (s.compressed!=0 | s.verbose!=0 |\r
+ s.timefield!=0 | s.extension!=0)) {\r
+ print_usage_exit("illegal argument combination with -s");\r
+ }\r
+ \r
+ if (!s.extension) {\r
+ s.extension = (s.compressed) ? ".txt.gz" : ".txt";\r
+ }\r
+ \r
+ argc -= optind;\r
+ argv += optind;\r
+ if (argc<3) print_usage_exit("must specify hub info and query");\r
+ \r
+ if (sscanf(argv[0],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(gshub.port))!= 5 ) {\r
+ gslog(LOG_EMERG,"HUB IP NOT DEFINED");\r
+ exit(1);\r
+ }\r
+ gshub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);\r
+ gshub.port=htons(gshub.port);\r
+ instance_name=strdup(argv[1]);\r
+ if (set_hub(gshub)!=0) {\r
+ gslog(LOG_EMERG,"Could not set hub");\r
+ exit(1);\r
+ }\r
+ if (set_instance_name(instance_name)!=0) {\r
+ gslog(LOG_EMERG,"Could not set instance name");\r
+ exit(1);\r
+ }\r
+ \r
+ if (get_initinstance(gshub,instance_name,&dummyep,1)!=0) {\r
+ gslog(LOG_EMERG,"Did not receive signal that GS is initiated\n");\r
+ }\r
+ \r
+ \r
+ argc -=2;\r
+ argv +=2;\r
+ \r
+ s.query = argv[0];\r
+ \r
+ init(argc, argv, &s);\r
+ \r
+ process_data(&s);\r
+ \r
+ gslog(LOG_EMERG,"%s::internal error reached unexpected end", me);\r
+}\r
+\r
Code Review / com / gs-lite.git / blobdiff