-/* ------------------------------------------------
- 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 <stdlib.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 <schemaparser.h>
-#include "gsconfig.h"
-#include "gstypes.h"
-#include "gshub.h"
-
-#define MAXLINE 100000
-static unsigned tcpport=0;
-static char linebuf[MAXLINE];
-int listensockfd=0;
-int fd=0;
-gs_int32_t verbose=0;
-endpoint hub;
-endpoint ds;
-gs_sp_t sink_name;
-
-
-static void wait_for_feed() {
- struct sockaddr_in serv_addr,cli_addr;
- socklen_t clilen;
- struct sockaddr_in sin;
- socklen_t sin_sz;
- if (listensockfd==0) {
- gs_int32_t on = 1;
-
- if (verbose==2) {
- fprintf(stderr,"Create listen socket for port %u\n",tcpport);
- }
- listensockfd=socket(AF_INET, SOCK_STREAM, 0);
- if (listensockfd < 0) {
- gslog(LOG_EMERG,"Error:Could not create socket for tcp data stream");
- exit(1);
- }
- bzero((char *) &serv_addr, sizeof(serv_addr));
- serv_addr.sin_family = AF_INET;
- serv_addr.sin_addr.s_addr = INADDR_ANY;
- serv_addr.sin_port = htons(tcpport);
-#ifndef __linux__
- /* make sure we can reuse the common port rapidly */
- if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEPORT,
- (gs_sp_t )&on, sizeof(on)) != 0) {
- gslog(LOG_EMERG,"Error::could not set socket option\n");
- exit(1);
- }
-#endif
- if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEADDR,
- (gs_sp_t )&on, sizeof(on)) != 0) {
- gslog(LOG_EMERG,"Error::could not set socket option\n");
- exit(1);
- }
-
- if (bind(listensockfd, (struct sockaddr *) &serv_addr,
- sizeof(serv_addr)) < 0) {
- gslog(LOG_EMERG,"Error:Could not bind socket for tcp data stream");
- exit(1);
- }
- }
-
- if (verbose==2) {
- fprintf(stderr,"Socket created waiting for data producer\n");
- }
- if (listen(listensockfd,5)< 0) {
- fprintf(stderr,"Error::could not listen to socket for port %u \n",ntohs(serv_addr.sin_port));
- close(listensockfd);
- exit(1);
- }
- sin_sz=sizeof(sin);
- if (getsockname(listensockfd, (struct sockaddr *) &sin, &sin_sz) < 0) {
- fprintf(stderr,"Error::could not get local port number of listen socket\n");
- exit(1);
- }
- ds.ip=htonl(127<<24|1);
- ds.port=sin.sin_port;
- if (set_streamsink(hub,sink_name,ds)!=0) {
- fprintf(stderr,"Error::could not set sink in GSHUB for %s source name\n",sink_name);
- exit(1);
- }
-
- do {
- clilen = sizeof(cli_addr);
- fd=accept(listensockfd, (struct sockaddr *) &cli_addr, &clilen);
- if (fd<0) {
- gslog(LOG_EMERG,"Error:Could not accept connection on tcp socket");
- }
- } while (fd==0);
- if (verbose) {
- fprintf(stderr,"Producer found ready to rock!\n");
- }
-
-}
-
-
-static void gs_read(gs_sp_t buffer, gs_uint32_t length){
- gs_uint32_t used=0;
- gs_uint32_t cur;
-
- while(used < length) {
- if (verbose==2) {
- fprintf(stderr,"\tread %u out of %u\n",used,length);
- }
- if ((cur=read(fd,&(buffer[used]),length-used))<=0) {
- gslog(LOG_EMERG,"ERROR:could not read data from gdat stream");
- exit(0);
- }
- used+=cur;
- }
-}
-
-static void gs_read_line(gs_sp_t buffer, gs_uint32_t length){
- gs_uint32_t used=0;
- gs_uint32_t cur;
-
- while((used < (length-1)) && ((used==0)|| (buffer[used-1]!='\n'))) {
- if ((cur=read(fd,&(buffer[used]),1))<=0) {
- gslog(LOG_EMERG,"ERROR:could not read data from gdat stream");
- exit(0);
- }
- used+=cur;
- }
- buffer[used]=0;
-}
-
-static gs_uint32_t get_stream_tuple(gs_uint32_t * rsize,gs_sp_t rbuf,gs_uint32_t bufsz){
-
- gs_uint32_t nsz,sz;
- gs_read((gs_sp_t)&nsz,sizeof(gs_uint32_t));
- sz=ntohl(nsz);
- if (sz>MAXTUPLESZ) {
- gslog(LOG_EMERG,"Error::Illegal tuple received");
- exit(0);
- }
-
- if (sz>bufsz) {
- gslog(LOG_EMERG,"Error::Illegal tuple size received -- to long");
- exit(0);
- }
- *rsize=sz;
-
- gs_read(rbuf,(sz));
- return 0;
-}
-
-int main(int argc, char** argv) {
- gs_uint32_t streamid;
- gs_schemahandle_t schema;
-
- gs_uint32_t rstreamid;
- gs_uint32_t rsize;
- gs_int8_t rbuf[2*MAXTUPLESZ];
-
- gs_int32_t numberoffields;
- gs_int32_t y;
- gs_int32_t parserversion;
- gs_uint32_t schemalen;
- gs_sp_t asciischema;
- gs_sp_t me;
- gs_int32_t ch;
-
- gs_int8_t buf[1024];
- gs_uint32_t tip1,tip2,tip3,tip4;
-
- /* initialize host library and the sgroup */
-
- me=argv[0];
-
- while ((ch = getopt(argc, argv, "hvxp:")) != -1) {
- switch(ch) {
- case 'h':
- fprintf(stderr,"%s::usage: %s -v -x -p <port> <IP>:<port> <source_name> \n",argv[0],argv[0]);
- exit(0);
- break;
- case 'p':
- tcpport=atoi(optarg);;
- break;
- case 'v':
- verbose=1;
- break;
- case 'x':
- verbose=2;
- break;
- default:
- break;
- }
- }
- if (optind+2>argc) {
- fprintf(stderr,"Could not find hub and stream source name on command line\n");
- exit(1);
- }
- if (sscanf(argv[optind],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(hub.port))!= 5 ) {
- fprintf(stderr,"Could not parse hub endpoint\n");
- exit(1);
- }
- hub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
- hub.port=htons(hub.port);
- sink_name=strdup(argv[optind+1]);
-
- wait_for_feed();
-
- gs_read_line(buf,1024);
- if (strncmp(buf,"GDAT",4)!=0) {
- gslog(LOG_EMERG,"ERROR: not a GDAT stream\n");
- exit(0);
- }
- gs_read_line(buf,1024);
- if (sscanf(buf,"VERSION:%u\n",&parserversion)!=1) {
- gslog(LOG_EMERG,"ERROR: no GDAT VERSION given\n");
- exit(0);
- }
- gs_read_line(buf,1024);
- if (sscanf(buf,"SCHEMALENGTH:%u\n",&schemalen)!=1) {
- gslog(LOG_EMERG,"ERROR: no GDAT SCHEMALENGTH given\n");
- exit(0);
- }
-
- if(verbose==2) {
- fprintf (stderr,"GDAT version %u schemalength %u\n",parserversion,schemalen);
- }
-
- if (schemaparser_accepts_version(parserversion)!=1) {
- fprintf(stderr,"%s::error: wrong parser version %u for file %s\n",
- me,parserversion,argv[0]);
- exit(1);
- }
-
- if ((asciischema=malloc(schemalen))==0) {
- fprintf(stderr,"%s::error: could not allocate schema buffer of sz %u "
- "for file %s\n",
- me,schemalen,argv[0]);
- exit(1);
- }
-
- if(verbose==2) {
- fprintf (stderr,"Reading schema from scocket\n");
- }
-
- gs_read(asciischema,schemalen);
-
- if(verbose==2) {
- fprintf (stderr,"Received schema\n");
- }
-
- if (verbose==2) {
- fprintf(stderr,"%s\n",asciischema);
- }
- if ((schema=ftaschema_parse_string(asciischema))<0) {
- fprintf(stderr,"%s::error: could not parse schema \n",
- me);
- exit(1);
- }
-
- if ((numberoffields=ftaschema_tuple_len(schema))<0) {
- fprintf(stderr,"%s::error:could not get number of fields in schema\n",
- me);
- exit(1);
- }
- if (verbose==1) {
- for(y=0; y<numberoffields;y++) {
- printf("%s",ftaschema_field_name(schema,y));
- if (y<numberoffields-1) printf("|");
- }
- printf("\n");
- }
- while(get_stream_tuple(&rsize,rbuf,2*MAXTUPLESZ)==0) {
- if (verbose) {
- if (ftaschema_is_eof_tuple(schema, rbuf)) {
- /* initiate shutdown or something of that nature */
- fprintf(stderr,"%s::All data proccessed\n",me);
- } else {
- if (ftaschema_is_temporal_tuple(schema, rbuf)) {
- /* initiate shutdown or something of that nature */
- fprintf(stderr,"%s:: temporal tuple\n",me);
- } else {
- fprintf(stderr,"%s:: regular tuple\n",me);
- }
- }
- }
-
- if ((!rsize) )
- continue;
- for(y=0; y<numberoffields;y++) {
- struct access_result ar;
- if (verbose==2)
- printf("%s->",ftaschema_field_name(schema,y));
- ar=ftaschema_get_field_by_index(schema,y,rbuf,rsize);
- switch (ar.field_data_type) {
- case INT_TYPE:
- printf("%d",ar.r.i);
- break;
- case UINT_TYPE:
- printf("%u",ar.r.ui);
- break;
- case IP_TYPE:
- printf("%u.%u.%u.%u",ar.r.ui>>24&0xff,
- ar.r.ui>>16&0xff,
- ar.r.ui>>8&0xff,
- ar.r.ui&0xff);
- break;
- case IPV6_TYPE:
- {
- unsigned x;
- unsigned zc=0;
- for(x=0;x<4;x++) { if (ar.r.ip6.v[x]==0) zc++;}
- if (zc!=4) {
- for(x=0;x<8;x++) {
- unsigned char * a = (unsigned char *) &(ar.r.ip6.v[0]);
- unsigned y;
- y=((unsigned)a[2*x])<<8|((unsigned) a[2*x+1]);
- printf("%04x",y);
- if (x<7) printf(":");
- }
- } else {
- printf("::");
- }
- }
- break;
- case USHORT_TYPE:
- printf("%hu",(unsigned short)ar.r.ui);
- break;
- case BOOL_TYPE:
- if (ar.r.ui==0) {
- printf("FALSE");
- } else {
- printf("TRUE");
- }
- break;
- case ULLONG_TYPE:
- {
- gs_uint64_t ul;
- gs_uint64_t t1;
- gs_uint64_t t2;
- ul=ar.r.ul;
- printf("%llu",ul);
- }
- break;
- case LLONG_TYPE:
- printf("%lld",ar.r.l);
- break;
- case FLOAT_TYPE:
- printf("%f",ar.r.f);
- break;
- case TIMEVAL_TYPE:
- {
- gs_float_t t;
- t= ar.r.t.tv_usec;
- t=t/1000000;
- t=t+ar.r.t.tv_sec;
- printf("%lf sec",t);
- }
- break;
- case VSTR_TYPE:
- {
- int x;
- int c;
- char * src;
- src=(gs_sp_t)ar.r.vs.offset;
- if ((ar.r.vs.length>0) && (src[ar.r.vs.length-1]==0)) {
- ar.r.vs.length = ar.r.vs.length-1;
- }
- for(x=0;x<ar.r.vs.length;x++) {
- c=src[x];
- if (((c<='~') && (c>=' '))&&(c!='|')) {
- printf("%c",c);
- } else {
- printf("(0x%x)",(gs_uint8_t)c);
- }
- }
- }
- break;
- default:
- break;
- }
- if (y<numberoffields-1) printf("|");
- }
- printf("\n");
- if (verbose!=0) fflush(stdout);
- }
- exit(0);
-}
-
+/* ------------------------------------------------\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 <stdlib.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 <schemaparser.h>\r
+#include "gsconfig.h"\r
+#include "gstypes.h"\r
+#include "gshub.h"\r
+\r
+#define MAXLINE 100000\r
+static unsigned tcpport=0;\r
+static char linebuf[MAXLINE];\r
+int listensockfd=0;\r
+int fd=0;\r
+gs_int32_t verbose=0;\r
+endpoint hub;\r
+endpoint ds;\r
+gs_sp_t sink_name;\r
+\r
+\r
+static void wait_for_feed() {\r
+ struct sockaddr_in serv_addr,cli_addr;\r
+ socklen_t clilen;\r
+ struct sockaddr_in sin;\r
+ socklen_t sin_sz;\r
+ if (listensockfd==0) {\r
+ gs_int32_t on = 1;\r
+ \r
+ if (verbose==2) {\r
+ fprintf(stderr,"Create listen socket for port %u\n",tcpport);\r
+ }\r
+ listensockfd=socket(AF_INET, SOCK_STREAM, 0);\r
+ if (listensockfd < 0) {\r
+ gslog(LOG_EMERG,"Error:Could not create socket for tcp data stream");\r
+ exit(1);\r
+ }\r
+ bzero((char *) &serv_addr, sizeof(serv_addr));\r
+ serv_addr.sin_family = AF_INET;\r
+ serv_addr.sin_addr.s_addr = INADDR_ANY;\r
+ serv_addr.sin_port = htons(tcpport);\r
+#ifndef __linux__\r
+ /* make sure we can reuse the common port rapidly */\r
+ if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEPORT,\r
+ (gs_sp_t )&on, sizeof(on)) != 0) {\r
+ gslog(LOG_EMERG,"Error::could not set socket option\n");\r
+ exit(1);\r
+ }\r
+#endif\r
+ if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEADDR,\r
+ (gs_sp_t )&on, sizeof(on)) != 0) {\r
+ gslog(LOG_EMERG,"Error::could not set socket option\n");\r
+ exit(1);\r
+ }\r
+ \r
+ if (bind(listensockfd, (struct sockaddr *) &serv_addr,\r
+ sizeof(serv_addr)) < 0) {\r
+ gslog(LOG_EMERG,"Error:Could not bind socket for tcp data stream");\r
+ exit(1);\r
+ }\r
+ }\r
+ \r
+ if (verbose==2) {\r
+ fprintf(stderr,"Socket created waiting for data producer\n");\r
+ }\r
+ if (listen(listensockfd,5)< 0) {\r
+ fprintf(stderr,"Error::could not listen to socket for port %u \n",ntohs(serv_addr.sin_port));\r
+ close(listensockfd);\r
+ exit(1);\r
+ }\r
+ sin_sz=sizeof(sin);\r
+ if (getsockname(listensockfd, (struct sockaddr *) &sin, &sin_sz) < 0) {\r
+ fprintf(stderr,"Error::could not get local port number of listen socket\n");\r
+ exit(1);\r
+ }\r
+ ds.ip=htonl(127<<24|1);\r
+ ds.port=sin.sin_port;\r
+ if (set_streamsink(hub,sink_name,ds)!=0) {\r
+ fprintf(stderr,"Error::could not set sink in GSHUB for %s source name\n",sink_name);\r
+ exit(1);\r
+ }\r
+ \r
+ do {\r
+ clilen = sizeof(cli_addr);\r
+ fd=accept(listensockfd, (struct sockaddr *) &cli_addr, &clilen);\r
+ if (fd<0) {\r
+ gslog(LOG_EMERG,"Error:Could not accept connection on tcp socket");\r
+ }\r
+ } while (fd==0);\r
+ if (verbose) {\r
+ fprintf(stderr,"Producer found ready to rock!\n");\r
+ }\r
+ \r
+}\r
+\r
+\r
+static void gs_read(gs_sp_t buffer, gs_uint32_t length){\r
+ gs_uint32_t used=0;\r
+ gs_uint32_t cur;\r
+ \r
+ while(used < length) {\r
+ if (verbose==2) {\r
+ fprintf(stderr,"\tread %u out of %u\n",used,length);\r
+ }\r
+ if ((cur=read(fd,&(buffer[used]),length-used))<=0) {\r
+ gslog(LOG_EMERG,"ERROR:could not read data from gdat stream");\r
+ exit(0);\r
+ }\r
+ used+=cur;\r
+ }\r
+}\r
+\r
+static void gs_read_line(gs_sp_t buffer, gs_uint32_t length){\r
+ gs_uint32_t used=0;\r
+ gs_uint32_t cur;\r
+ \r
+ while((used < (length-1)) && ((used==0)|| (buffer[used-1]!='\n'))) {\r
+ if ((cur=read(fd,&(buffer[used]),1))<=0) {\r
+ gslog(LOG_EMERG,"ERROR:could not read data from gdat stream");\r
+ exit(0);\r
+ }\r
+ used+=cur;\r
+ }\r
+ buffer[used]=0;\r
+}\r
+\r
+static gs_uint32_t get_stream_tuple(gs_uint32_t * rsize,gs_sp_t rbuf,gs_uint32_t bufsz){\r
+ \r
+ gs_uint32_t nsz,sz;\r
+ gs_read((gs_sp_t)&nsz,sizeof(gs_uint32_t));\r
+ sz=ntohl(nsz);\r
+ if (sz>MAXTUPLESZ) {\r
+ gslog(LOG_EMERG,"Error::Illegal tuple received");\r
+ exit(0);\r
+ }\r
+ \r
+ if (sz>bufsz) {\r
+ gslog(LOG_EMERG,"Error::Illegal tuple size received -- to long");\r
+ exit(0);\r
+ }\r
+ *rsize=sz;\r
+ \r
+ gs_read(rbuf,(sz));\r
+ return 0;\r
+}\r
+\r
+int main(int argc, char** argv) {\r
+ gs_uint32_t streamid;\r
+ gs_schemahandle_t schema;\r
+ \r
+ gs_uint32_t rstreamid;\r
+ gs_uint32_t rsize;\r
+ gs_int8_t rbuf[2*MAXTUPLESZ];\r
+ \r
+ gs_int32_t numberoffields;\r
+ gs_int32_t y;\r
+ gs_int32_t parserversion;\r
+ gs_uint32_t schemalen;\r
+ gs_sp_t asciischema;\r
+ gs_sp_t me;\r
+ gs_int32_t ch;\r
+ \r
+ gs_int8_t buf[1024];\r
+ gs_uint32_t tip1,tip2,tip3,tip4;\r
+ \r
+ /* initialize host library and the sgroup */\r
+ \r
+ me=argv[0];\r
+ \r
+ while ((ch = getopt(argc, argv, "hvxp:")) != -1) {\r
+ switch(ch) {\r
+ case 'h':\r
+ fprintf(stderr,"%s::usage: %s -v -x -p <port> <IP>:<port> <source_name> \n",argv[0],argv[0]);\r
+ exit(0);\r
+ break;\r
+ case 'p':\r
+ tcpport=atoi(optarg);;\r
+ break;\r
+ case 'v':\r
+ verbose=1;\r
+ break;\r
+ case 'x':\r
+ verbose=2;\r
+ break;\r
+ default:\r
+ break;\r
+ }\r
+ }\r
+ if (optind+2>argc) {\r
+ fprintf(stderr,"Could not find hub and stream source name on command line\n");\r
+ exit(1);\r
+ }\r
+ if (sscanf(argv[optind],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(hub.port))!= 5 ) {\r
+ fprintf(stderr,"Could not parse hub endpoint\n");\r
+ exit(1);\r
+ }\r
+ hub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);\r
+ hub.port=htons(hub.port);\r
+ sink_name=strdup(argv[optind+1]);\r
+ \r
+ wait_for_feed();\r
+ \r
+ gs_read_line(buf,1024);\r
+ if (strncmp(buf,"GDAT",4)!=0) {\r
+ gslog(LOG_EMERG,"ERROR: not a GDAT stream\n");\r
+ exit(0);\r
+ }\r
+ gs_read_line(buf,1024);\r
+ if (sscanf(buf,"VERSION:%u\n",&parserversion)!=1) {\r
+ gslog(LOG_EMERG,"ERROR: no GDAT VERSION given\n");\r
+ exit(0);\r
+ }\r
+ gs_read_line(buf,1024);\r
+ if (sscanf(buf,"SCHEMALENGTH:%u\n",&schemalen)!=1) {\r
+ gslog(LOG_EMERG,"ERROR: no GDAT SCHEMALENGTH given\n");\r
+ exit(0);\r
+ }\r
+ \r
+ if(verbose==2) {\r
+ fprintf (stderr,"GDAT version %u schemalength %u\n",parserversion,schemalen);\r
+ }\r
+ \r
+ if (schemaparser_accepts_version(parserversion)!=1) {\r
+ fprintf(stderr,"%s::error: wrong parser version %u for file %s\n",\r
+ me,parserversion,argv[0]);\r
+ exit(1);\r
+ }\r
+ \r
+ if ((asciischema=malloc(schemalen))==0) {\r
+ fprintf(stderr,"%s::error: could not allocate schema buffer of sz %u "\r
+ "for file %s\n",\r
+ me,schemalen,argv[0]);\r
+ exit(1);\r
+ }\r
+ \r
+ if(verbose==2) {\r
+ fprintf (stderr,"Reading schema from scocket\n");\r
+ }\r
+ \r
+ gs_read(asciischema,schemalen);\r
+ \r
+ if(verbose==2) {\r
+ fprintf (stderr,"Received schema\n");\r
+ }\r
+ \r
+ if (verbose==2) {\r
+ fprintf(stderr,"%s\n",asciischema);\r
+ }\r
+ if ((schema=ftaschema_parse_string(asciischema))<0) {\r
+ fprintf(stderr,"%s::error: could not parse schema \n",\r
+ me);\r
+ exit(1);\r
+ }\r
+ \r
+ if ((numberoffields=ftaschema_tuple_len(schema))<0) {\r
+ fprintf(stderr,"%s::error:could not get number of fields in schema\n",\r
+ me);\r
+ exit(1);\r
+ }\r
+ if (verbose==1) {\r
+ for(y=0; y<numberoffields;y++) {\r
+ printf("%s",ftaschema_field_name(schema,y));\r
+ if (y<numberoffields-1) printf("|");\r
+ }\r
+ printf("\n");\r
+ }\r
+ while(get_stream_tuple(&rsize,rbuf,2*MAXTUPLESZ)==0) {\r
+ if (verbose) {\r
+ if (ftaschema_is_eof_tuple(schema, rbuf)) {\r
+ /* initiate shutdown or something of that nature */\r
+ fprintf(stderr,"%s::All data proccessed\n",me);\r
+ } else {\r
+ if (ftaschema_is_temporal_tuple(schema, rbuf)) {\r
+ /* initiate shutdown or something of that nature */\r
+ fprintf(stderr,"%s:: temporal tuple\n",me);\r
+ } else {\r
+ fprintf(stderr,"%s:: regular tuple\n",me);\r
+ }\r
+ }\r
+ }\r
+ \r
+ if ((!rsize) )\r
+ continue;\r
+ for(y=0; y<numberoffields;y++) {\r
+ struct access_result ar;\r
+ if (verbose==2)\r
+ printf("%s->",ftaschema_field_name(schema,y));\r
+ ar=ftaschema_get_field_by_index(schema,y,rbuf,rsize);\r
+ switch (ar.field_data_type) {\r
+ case INT_TYPE:\r
+ printf("%d",ar.r.i);\r
+ break;\r
+ case UINT_TYPE:\r
+ printf("%u",ar.r.ui);\r
+ break;\r
+ case IP_TYPE:\r
+ printf("%u.%u.%u.%u",ar.r.ui>>24&0xff,\r
+ ar.r.ui>>16&0xff,\r
+ ar.r.ui>>8&0xff,\r
+ ar.r.ui&0xff);\r
+ break;\r
+ case IPV6_TYPE:\r
+ {\r
+ unsigned x;\r
+ unsigned zc=0;\r
+ for(x=0;x<4;x++) { if (ar.r.ip6.v[x]==0) zc++;}\r
+ if (zc!=4) {\r
+ for(x=0;x<8;x++) {\r
+ unsigned char * a = (unsigned char *) &(ar.r.ip6.v[0]);\r
+ unsigned y;\r
+ y=((unsigned)a[2*x])<<8|((unsigned) a[2*x+1]);\r
+ printf("%04x",y);\r
+ if (x<7) printf(":");\r
+ }\r
+ } else {\r
+ printf("::");\r
+ }\r
+ }\r
+ break;\r
+ case USHORT_TYPE:\r
+ printf("%hu",(unsigned short)ar.r.ui);\r
+ break;\r
+ case BOOL_TYPE:\r
+ if (ar.r.ui==0) {\r
+ printf("FALSE");\r
+ } else {\r
+ printf("TRUE");\r
+ }\r
+ break;\r
+ case ULLONG_TYPE:\r
+ {\r
+ gs_uint64_t ul;\r
+ gs_uint64_t t1;\r
+ gs_uint64_t t2;\r
+ ul=ar.r.ul;\r
+ printf("%llu",ul);\r
+ }\r
+ break;\r
+ case LLONG_TYPE:\r
+ printf("%lld",ar.r.l);\r
+ break;\r
+ case FLOAT_TYPE:\r
+ printf("%f",ar.r.f);\r
+ break;\r
+ case TIMEVAL_TYPE:\r
+ { \r
+ gs_float_t t;\r
+ t= ar.r.t.tv_usec;\r
+ t=t/1000000;\r
+ t=t+ar.r.t.tv_sec;\r
+ printf("%lf sec",t);\r
+ }\r
+ break;\r
+ case VSTR_TYPE:\r
+ {\r
+ int x;\r
+ int c;\r
+ char * src;\r
+ src=(gs_sp_t)ar.r.vs.offset;\r
+ if ((ar.r.vs.length>0) && (src[ar.r.vs.length-1]==0)) {\r
+ ar.r.vs.length = ar.r.vs.length-1;\r
+ }\r
+ for(x=0;x<ar.r.vs.length;x++) {\r
+ c=src[x];\r
+ if (((c<='~') && (c>=' '))&&(c!='|')) {\r
+ printf("%c",c);\r
+ } else {\r
+ printf("(0x%x)",(gs_uint8_t)c);\r
+ }\r
+ }\r
+ }\r
+ break;\r
+ default:\r
+ break;\r
+ }\r
+ if (y<numberoffields-1) printf("|");\r
+ }\r
+ printf("\n");\r
+ if (verbose!=0) fflush(stdout);\r
+ }\r
+ exit(0);\r
+}\r
+\r
Code Review / com / gs-lite.git / blobdiff