Fixed newline characters throughout the code
[com/gs-lite.git] / src / tools / gssinksim.c
index 89a5f18..9334ee8 100644 (file)
-/* ------------------------------------------------\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
+/* ------------------------------------------------
+ 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);
+}
+