X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Ftools%2Fgsprintconsole_ves.c;h=5773c9fe2bd3204a8950a07aee3c2251c6a8411a;hb=07495effe193ca3f73c3bf0ce417068f9ac9dcdd;hp=30ddab8da68691396244b978248319fbf885849a;hpb=93d248304a68de7a8f9daf4aa74f9ee4cd27410c;p=com%2Fgs-lite.git

diff --git a/src/tools/gsprintconsole_ves.c b/src/tools/gsprintconsole_ves.c
index 30ddab8..5773c9f 100644
--- a/src/tools/gsprintconsole_ves.c
+++ b/src/tools/gsprintconsole_ves.c
@@ -1,643 +1,643 @@
-/* ------------------------------------------------
- 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.
- ------------------------------------------- */
-
-
-/*
- * Print ves formatted records to the console.
- * Each line is a json record.
- * Based on gsprintconsole.c, just differences in formatting.
-*/
-
-
-#include <app.h>
-#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 "gsconfig.h"
-#include "gstypes.h"
-#include "gshub.h"
-#include "simple_http.h"
-
-#include <schemaparser.h>
-
-#define MAXLINE 100000
-static unsigned tcpport=0;
-static char linebuf[MAXLINE];
-int listensockfd=0;
-int fd=0;
-
-
-// Not all systems have timersub defined so make sure its ther
-#ifndef timersub
-
-#define timersub(tvp, uvp, vvp)                                         \
-do {                                                            \
-(vvp)->tv_sec = (tvp)->tv_sec - (uvp)->tv_sec;          \
-(vvp)->tv_usec = (tvp)->tv_usec - (uvp)->tv_usec;       \
-if ((vvp)->tv_usec < 0) {                               \
-(vvp)->tv_sec--;                                \
-(vvp)->tv_usec += 1000000;                      \
-}                                                       \
-} while (0)
-
-#endif
-
-void hand(int iv) {
-    ftaapp_exit();
-    fprintf(stderr, "exiting via signal handler %d...\n", iv);
-    exit(1);
-}
-
-static void wait_for_client() {
-    struct sockaddr_in serv_addr,cli_addr;
-    socklen_t clilen;
-    if (listensockfd==0) {
-		gs_int32_t on = 1;
-		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);
-        }
-	}
-    
-	do {
-		listen(listensockfd,5);
-		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);
-}
-
-
-static void emit_socket() {
-	unsigned o,w,l;
-	o=0;
-	w=0;
-	l=strlen(linebuf);
-	do {
-		if((w=write(fd,&linebuf[o],l))==0) {
-			close(fd);
-			wait_for_client();
-		}
-		o=o+w;
-	} while (o<l);
-}
-
-static void emit_line() {
-    
-    if (tcpport==0) {
-        printf("%s",linebuf);
-    } else {
-        emit_socket();
-    }
-    
-}
-
-int main(int argc, char* argv[]) {
-    gs_sp_t me = argv[0];
-    FTAID fta_id;
-    gs_int32_t schema, ch;
-    
-    FTAID rfta_id;
-    gs_uint32_t rsize;
-    gs_uint32_t bufsz=8*1024*1024;
-    gs_int8_t rbuf[2*MAXTUPLESZ];
-    
-    gs_int32_t numberoffields;
-    gs_int32_t verbose=0;
-    gs_int32_t y, lcv;
-    
-    void *pblk;
-    gs_int32_t pblklen;
-	gs_int32_t n_actual_param;
-	gs_int32_t n_expected_param;
-    gs_int32_t xit = 0;
-    gs_int32_t dump = 0;
-    struct timeval tvs, tve, tvd;
-    gs_retval_t code;
-    endpoint gshub;
-    endpoint dummyep;
-    gs_uint32_t tip1,tip2,tip3,tip4;
-    gs_sp_t instance_name;
-
-	gs_sp_t curl_address = NULL;
-	endpoint curl_endpoint;
-	gs_sp_t curl_url = NULL;
-	gs_sp_t curl_auth = NULL;
-	gs_uint32_t http_code;
-
-	gs_uint32_t ves_version=7;
-
-
-    gs_uint32_t tlimit = 0;     // time limit in seconds
-    time_t start_time, curr_time;
-    
-	gsopenlog(argv[0]);
-    
-    
-    while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:A:V:")) != -1) {
-        switch (ch) {
-            case 'r':
-                bufsz=atoi(optarg);
-                break;
-            case 'p':
-                tcpport=atoi(optarg);
-                break;
-            case 'v':
-                verbose++;
-                break;
-            case 'X':
-                xit++;
-                break;
-            case 'D':
-                dump++;
-                break;
-            case 'l':
-                tlimit = atoi(optarg);
-                break;
-            case 'V':
-                ves_version = atoi(optarg);
-                break;
-			case 'C':
-				curl_address = strdup(optarg);
-    			if (sscanf(curl_address,"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(curl_endpoint.port))!= 5 ) {
-        			gslog(LOG_EMERG,"Curl IP NOT DEFINED");
-        			exit(1);
-    			}
-    			curl_endpoint.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
-    			curl_endpoint.port=htons(curl_endpoint.port);
-				break;
-            case 'U':
-				curl_url = strdup(optarg);
-                break;
-            case 'A':
-				curl_auth = strdup(optarg);
-                break;
-            default:
-            usage:
-                fprintf(stderr, "usage: %s [-r <bufsz>] [-p <port>] [-l <time_limit>] [-v] [-X] [-D] [-C <curl_dest>:<curl_port>] [-U <curl_url>] [-A <authentication_string>] [-V <ves_version>] <gshub-hostname>:<gshub-port> <gsinstance_name>  query param1 param2...\n",
-                        *argv);
-                exit(1);
-        }
-    }
-    argc -= optind;
-    argv += optind;
-    if (argc<3) goto usage;
-    
-    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");
-    }
-
-
-//	If this uses curl output, ensure consistency in the curl args
-	if(curl_address != NULL){
-		if(curl_url == NULL){
-			gslog(LOG_EMERG,"Curl IP defined, but there is no url (e.g. /foo/bar");
-			exit(1);
-		}
-		if(curl_auth==NULL){
-			curl_auth = "";
-		} 
-	}
-    
-    gettimeofday(&tvs, 0);
-    argc -=2;
-    argv +=2;
-    if (argc < 1)
-        goto usage;
-    
-    /* initialize host library and the sgroup  */
-    
-    if (verbose>=2) fprintf(stderr,"Initializing gscp\n");
-    
-    if (ftaapp_init(bufsz)!=0) {
-        fprintf(stderr,"%s::error:could not initialize gscp\n", me);
-        exit(1);
-    }
-    
-    signal(SIGTERM, hand);
-    signal(SIGINT, hand);
-    
-    schema = ftaapp_get_fta_schema_by_name(argv[0]);
-    if (schema < 0) {
-        fprintf(stderr,"%s::error:could not 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;
-    } else {
-		n_actual_param = argc-1;
-		if(n_actual_param < n_expected_param){
-			fprintf(stderr,"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) {
-                fprintf(stderr,"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) {
-                fprintf(stderr,"param setparam error '%s' (fmt 'key=val')\n",
-                        argv[lcv]);
-                exit(1);
-            }
-        }
-        if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {
-            fprintf(stderr, "ftaschema_create_param_block failed!\n");
-            exit(1);
-        }
-    }
-//    ftaschema_free(schema); /* XXXCDC */ // the schema continues to be used
-    
-    
-    if (verbose>=2) fprintf(stderr,"Initalized FTA\n");
-    
-    fta_id=ftaapp_add_fta(argv[0],0,0,0,pblklen,pblk);
-    if (fta_id.streamid==0) {
-        fprintf(stderr,"%s::error:could not initialize fta %s\n",
-                me, argv[0]);
-        exit(1);
-    }
-    /* XXXCDC: pblk is malloc'd, should we free it? */
-    
-    if (verbose>=2) fprintf(stderr,"Get schema handle\n");
-    
-    if ((schema=ftaapp_get_fta_schema(fta_id))<0) {
-        fprintf(stderr,"%s::error:could not get 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");
-    }
-    if (xit) {	// -X in command line
-        gettimeofday(&tve, 0);
-        timersub(&tve, &tvs, &tvd);
-        printf("TIME= %ld%06d sec\n", tvd.tv_sec, tvd.tv_usec);
-        hand(0);	// effectively an exit
-    }
-    if (tcpport!=0) {
-    	wait_for_client();
-    }
-
-    start_time = time(NULL);
-
-	int measurement_interval_pos = -1; // extract measurementInterval if present
-	char *field_names[numberoffields];
-    for(y=0; y<numberoffields;y++) {
-		field_names[y] = strdup(ftaschema_field_name(schema,y));
-		if(strcmp(field_names[y], "measurementInterval")==0)
-			measurement_interval_pos = y;
-	}
-
-
-	struct timeval tsample;
-	gettimeofday(&tsample, 0);
-	char start_ts[100], curr_ts[100];
-	sprintf(start_ts,"%ld%06d", tsample.tv_sec, tsample.tv_usec);
-
-	long unsigned int lineno=0;
-	long unsigned int seqno=0;
-	unsigned int measurement_interval;
-    while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {
-		lineno++;
-        if (dump)	// -D in command line
-            continue;
-        if (ftaschema_is_eof_tuple(schema, rbuf)) {
-            /* initiate shutdown or something of that nature */
-            printf("#All data proccessed\n");
-            exit(0);
-        }
-        if (!rsize)
-            continue;
-        if (verbose >=2) {
-            snprintf(linebuf,MAXLINE,"RESULT CODE => %u\n",code);
-            emit_line();
-        }
-        if ((code==0)&&(rfta_id.streamid == fta_id.streamid)) {
-			seqno++;
-			gettimeofday(&tsample, 0);
-			sprintf(curr_ts,"%ld%06d", tsample.tv_sec, tsample.tv_usec);
-			int pos;
-			if(ves_version < 7){
-				pos = snprintf(linebuf, MAXLINE,
-  "{\"event\": { \"commonEventHeader\": { "
-        "\"domain\": \"measurementsForVfScaling\", "
-        "\"eventId\": \"%s%u\", "
-        "\"eventType\": \"%s\", "
-        "\"eventName\": \"Measurement_MC_%s\", "
-        "\"lastEpochMicrosec\": %s, "
-        "\"priority\": \"Normal\", "
-        "\"reportingEntityName\": \"GS-LITE MC\", "
-        "\"sequence\": %u, "
-        "\"sourceName\": \"meas_cmpgn_xapp\", "
-        "\"startEpochMicrosec\": %s, "
-        "\"version\": 5 "
-      "}, "
-      "\"measurementsForVfScalingFields\": { "
-	  	"\"additionalFields\": ["
-				,argv[0],lineno, argv[0], argv[0], curr_ts, seqno, start_ts
-				);
-			}else{
-				pos = snprintf(linebuf, MAXLINE,
-  "{\"event\": { \"commonEventHeader\": { "
-        "\"domain\": \"measurement\", "
-        "\"eventId\": \"%s%u\", "
-        "\"eventType\": \"%s\", "
-        "\"eventName\": \"Measurement_MC_%s\", "
-        "\"lastEpochMicrosec\": %s, "
-        "\"priority\": \"Normal\", "
-        "\"reportingEntityName\": \"GS-LITE MC\", "
-        "\"sequence\": %u, "
-        "\"sourceName\": \"meas_cmpgn_xapp\", "
-        "\"startEpochMicrosec\": %s, "
-        "\"version\": \"4.0.1\", "
-        "\"vesEventListenerVersion\": \"7.0.1\" "
-      "}, "
-      "\"measurementFields\": { "
-	  	"\"additionalFields\": {"
-				,argv[0],lineno, argv[0], argv[0], curr_ts, seqno, start_ts
-				);
-			}
-
-			measurement_interval = 0;
-            for(y=0; y<numberoffields;y++) {
-                struct access_result ar;
-//                if (verbose>=2)
-//                    printf("%s->",ftaschema_field_name(schema,y));
-              	if(y>0){
-					linebuf[pos]=',';
-					pos++;
-				}
-                ar=ftaschema_get_field_by_index(schema,y,rbuf,rsize);
-                switch (ar.field_data_type) {
-                    case INT_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%d\"}",field_names[y], ar.r.i);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%d\"",field_names[y], ar.r.i);
-						if(y==measurement_interval_pos)
-							measurement_interval = (unsigned int)ar.r.i;
-                        break;
-                    case UINT_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u\"}",field_names[y], ar.r.ui);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%u\"",field_names[y], ar.r.ui);
-						if(y==measurement_interval_pos)
-							measurement_interval = (unsigned int)ar.r.ui;
-                        break;
-                    case IP_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u.%u.%u.%u\"}",field_names[y], ar.r.ui>>24&0xff,
-                                 ar.r.ui>>16&0xff,
-                                 ar.r.ui>>8&0xff,
-                                 ar.r.ui&0xff);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%u.%u.%u.%u\"",field_names[y], ar.r.ui>>24&0xff,
-                                 ar.r.ui>>16&0xff,
-                                 ar.r.ui>>8&0xff,
-                                 ar.r.ui&0xff);
-                        break;
-                    case IPV6_TYPE:
-                    {
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"",field_names[y]);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"",field_names[y]);
-                        unsigned x;
-                        unsigned zc=0;
-                        for(x=0;x<4;x++) { if (ar.r.ip6.v[x]==0) zc++;}
-                        if (zc!=4) {
-                            snprintf(linebuf,MAXLINE,"");
-                            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]);
-                                pos += snprintf(linebuf+pos,MAXLINE-pos,"%04x",y);
-                                if (x<7){
-									pos += snprintf(linebuf+pos,MAXLINE-pos,":");
-								}
-                            }
-                        } else {
-                            pos+=snprintf(linebuf+pos,MAXLINE-pos,"::");
-                        }
-						if(ves_version < 7)
-							pos += snprintf(linebuf+pos, MAXLINE-pos,"\"}");
-						else
-							pos += snprintf(linebuf+pos, MAXLINE-pos,"\"");
-                    }
-                        break;
-                        
-                    case USHORT_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u\"}",field_names[y], ar.r.ui);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\":  \"%u\"}",field_names[y], ar.r.ui);
-						if(y==measurement_interval_pos)
-							measurement_interval = (unsigned int)ar.r.ui;
-                        break;
-                    case BOOL_TYPE:
-						if(ves_version < 7){
-	                        if (ar.r.ui==0) {
-   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"FALSE\"}",field_names[y]);
-   		                     } else {
-   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"TRUE\"}",field_names[y]);
-   	                     	}
-						}else{
-	                        if (ar.r.ui==0) {
-   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"FALSE\"",field_names[y]);
-   		                     } else {
-   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"TRUE\"",field_names[y]);
-   	                     	}
-						}
-                        break;
-                    case ULLONG_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%llu\"}",field_names[y], ar.r.ul);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%llu\"",field_names[y], ar.r.ul);
-						if(y==measurement_interval_pos)
-							measurement_interval = (unsigned int)ar.r.ul;
-                        break;
-                    case LLONG_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%lld\"}",field_names[y], ar.r.l);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%lld\"",field_names[y], ar.r.l);
-						if(y==measurement_interval_pos)
-							measurement_interval = (unsigned int)ar.r.l;
-                        break;
-                    case FLOAT_TYPE:
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%f\"}",field_names[y], ar.r.f);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%f\"",field_names[y], 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;
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%f\"}",field_names[y], t);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%f\"",field_names[y], t);
-                    }
-                        break;
-                    case VSTR_TYPE:
-                    {
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"",field_names[y]);
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"",field_names[y]);
-                        int x;
-                        int c;
-                        char * src;
-                        src=(char*)ar.r.vs.offset;
-						for(x=0;x<ar.r.vs.length;x++){
-                            c=src[x];
-                            if ((c<='~') && (c>=' ')) {
-                                if (pos<MAXLINE-1) {
-                                    linebuf[pos]=c;
-                                    pos++;
-                                }
-                            } else {
-                                if (pos<MAXLINE-1) {
-                                    linebuf[pos]='.';
-                                    pos++;
-                                }
-                            }
-						}
-						if(ves_version < 7)
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"}");
-						else
-                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"");
-                    }
-                        break;
-                    default:
-                        linebuf[0]=0;
-                        break;
-                }
-			}
-			if(ves_version < 7){
-				snprintf(linebuf+pos, MAXLINE-pos,
-	  	"], \"measurementInterval\": %u, \"measurementsForVfScalingVersion\": 1"
-		"}}}\n", measurement_interval
-				);
-			}else{
-				snprintf(linebuf+pos, MAXLINE-pos,
-	  	"}, \"measurementInterval\": %u, \"measurementFieldsVersion\": \"4.0\""
-		"}}}\n", measurement_interval
-				);
-			}
-			if(curl_address==NULL){
-            	emit_line();
-			}else{
-				http_post_request_hdr(curl_endpoint, curl_url, linebuf, &http_code, curl_auth);
-				if(http_code != 200 && http_code != 202){
-					gslog(LOG_WARNING, "http return code is %d\n",http_code);
-				}
-			}
-            if (verbose!=0) fflush(stdout);
-        } else {
-            if (rfta_id.streamid != fta_id.streamid)
-                fprintf(stderr,"Got unkown streamid %llu \n",rfta_id.streamid);
-        }
-
-        // whenever we receive a temp tuple check if we reached time limit
-        if ((code==2)  && tlimit && (time(NULL)-start_time)>=tlimit) {
-            fprintf(stderr,"Reached time limit of %d seconds\n",tlimit);    
-            ftaapp_exit();
-            exit(0);
-        }        
-    }
-}
-
+/* ------------------------------------------------
+ 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.
+ ------------------------------------------- */
+
+
+/*
+ * Print ves formatted records to the console.
+ * Each line is a json record.
+ * Based on gsprintconsole.c, just differences in formatting.
+*/
+
+
+#include <app.h>
+#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 "gsconfig.h"
+#include "gstypes.h"
+#include "gshub.h"
+#include "simple_http.h"
+
+#include <schemaparser.h>
+
+#define MAXLINE 100000
+static unsigned tcpport=0;
+static char linebuf[MAXLINE];
+int listensockfd=0;
+int fd=0;
+
+
+// Not all systems have timersub defined so make sure its ther
+#ifndef timersub
+
+#define timersub(tvp, uvp, vvp)                                         \
+do {                                                            \
+(vvp)->tv_sec = (tvp)->tv_sec - (uvp)->tv_sec;          \
+(vvp)->tv_usec = (tvp)->tv_usec - (uvp)->tv_usec;       \
+if ((vvp)->tv_usec < 0) {                               \
+(vvp)->tv_sec--;                                \
+(vvp)->tv_usec += 1000000;                      \
+}                                                       \
+} while (0)
+
+#endif
+
+void hand(int iv) {
+    ftaapp_exit();
+    fprintf(stderr, "exiting via signal handler %d...\n", iv);
+    exit(1);
+}
+
+static void wait_for_client() {
+    struct sockaddr_in serv_addr,cli_addr;
+    socklen_t clilen;
+    if (listensockfd==0) {
+		gs_int32_t on = 1;
+		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);
+        }
+	}
+    
+	do {
+		listen(listensockfd,5);
+		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);
+}
+
+
+static void emit_socket() {
+	unsigned o,w,l;
+	o=0;
+	w=0;
+	l=strlen(linebuf);
+	do {
+		if((w=write(fd,&linebuf[o],l))==0) {
+			close(fd);
+			wait_for_client();
+		}
+		o=o+w;
+	} while (o<l);
+}
+
+static void emit_line() {
+    
+    if (tcpport==0) {
+        printf("%s",linebuf);
+    } else {
+        emit_socket();
+    }
+    
+}
+
+int main(int argc, char* argv[]) {
+    gs_sp_t me = argv[0];
+    FTAID fta_id;
+    gs_int32_t schema, ch;
+    
+    FTAID rfta_id;
+    gs_uint32_t rsize;
+    gs_uint32_t bufsz=8*1024*1024;
+    gs_int8_t rbuf[2*MAXTUPLESZ];
+    
+    gs_int32_t numberoffields;
+    gs_int32_t verbose=0;
+    gs_int32_t y, lcv;
+    
+    void *pblk;
+    gs_int32_t pblklen;
+	gs_int32_t n_actual_param;
+	gs_int32_t n_expected_param;
+    gs_int32_t xit = 0;
+    gs_int32_t dump = 0;
+    struct timeval tvs, tve, tvd;
+    gs_retval_t code;
+    endpoint gshub;
+    endpoint dummyep;
+    gs_uint32_t tip1,tip2,tip3,tip4;
+    gs_sp_t instance_name;
+
+	gs_sp_t curl_address = NULL;
+	endpoint curl_endpoint;
+	gs_sp_t curl_url = NULL;
+	gs_sp_t curl_auth = NULL;
+	gs_uint32_t http_code;
+
+	gs_uint32_t ves_version=7;
+
+
+    gs_uint32_t tlimit = 0;     // time limit in seconds
+    time_t start_time, curr_time;
+    
+	gsopenlog(argv[0]);
+    
+    
+    while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:A:V:")) != -1) {
+        switch (ch) {
+            case 'r':
+                bufsz=atoi(optarg);
+                break;
+            case 'p':
+                tcpport=atoi(optarg);
+                break;
+            case 'v':
+                verbose++;
+                break;
+            case 'X':
+                xit++;
+                break;
+            case 'D':
+                dump++;
+                break;
+            case 'l':
+                tlimit = atoi(optarg);
+                break;
+            case 'V':
+                ves_version = atoi(optarg);
+                break;
+			case 'C':
+				curl_address = strdup(optarg);
+    			if (sscanf(curl_address,"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(curl_endpoint.port))!= 5 ) {
+        			gslog(LOG_EMERG,"Curl IP NOT DEFINED");
+        			exit(1);
+    			}
+    			curl_endpoint.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
+    			curl_endpoint.port=htons(curl_endpoint.port);
+				break;
+            case 'U':
+				curl_url = strdup(optarg);
+                break;
+            case 'A':
+				curl_auth = strdup(optarg);
+                break;
+            default:
+            usage:
+                fprintf(stderr, "usage: %s [-r <bufsz>] [-p <port>] [-l <time_limit>] [-v] [-X] [-D] [-C <curl_dest>:<curl_port>] [-U <curl_url>] [-A <authentication_string>] [-V <ves_version>] <gshub-hostname>:<gshub-port> <gsinstance_name>  query param1 param2...\n",
+                        *argv);
+                exit(1);
+        }
+    }
+    argc -= optind;
+    argv += optind;
+    if (argc<3) goto usage;
+    
+    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");
+    }
+
+
+//	If this uses curl output, ensure consistency in the curl args
+	if(curl_address != NULL){
+		if(curl_url == NULL){
+			gslog(LOG_EMERG,"Curl IP defined, but there is no url (e.g. /foo/bar");
+			exit(1);
+		}
+		if(curl_auth==NULL){
+			curl_auth = "";
+		} 
+	}
+    
+    gettimeofday(&tvs, 0);
+    argc -=2;
+    argv +=2;
+    if (argc < 1)
+        goto usage;
+    
+    /* initialize host library and the sgroup  */
+    
+    if (verbose>=2) fprintf(stderr,"Initializing gscp\n");
+    
+    if (ftaapp_init(bufsz)!=0) {
+        fprintf(stderr,"%s::error:could not initialize gscp\n", me);
+        exit(1);
+    }
+    
+    signal(SIGTERM, hand);
+    signal(SIGINT, hand);
+    
+    schema = ftaapp_get_fta_schema_by_name(argv[0]);
+    if (schema < 0) {
+        fprintf(stderr,"%s::error:could not 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;
+    } else {
+		n_actual_param = argc-1;
+		if(n_actual_param < n_expected_param){
+			fprintf(stderr,"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) {
+                fprintf(stderr,"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) {
+                fprintf(stderr,"param setparam error '%s' (fmt 'key=val')\n",
+                        argv[lcv]);
+                exit(1);
+            }
+        }
+        if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {
+            fprintf(stderr, "ftaschema_create_param_block failed!\n");
+            exit(1);
+        }
+    }
+//    ftaschema_free(schema); /* XXXCDC */ // the schema continues to be used
+    
+    
+    if (verbose>=2) fprintf(stderr,"Initalized FTA\n");
+    
+    fta_id=ftaapp_add_fta(argv[0],0,0,0,pblklen,pblk);
+    if (fta_id.streamid==0) {
+        fprintf(stderr,"%s::error:could not initialize fta %s\n",
+                me, argv[0]);
+        exit(1);
+    }
+    /* XXXCDC: pblk is malloc'd, should we free it? */
+    
+    if (verbose>=2) fprintf(stderr,"Get schema handle\n");
+    
+    if ((schema=ftaapp_get_fta_schema(fta_id))<0) {
+        fprintf(stderr,"%s::error:could not get 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");
+    }
+    if (xit) {	// -X in command line
+        gettimeofday(&tve, 0);
+        timersub(&tve, &tvs, &tvd);
+        printf("TIME= %ld%06d sec\n", tvd.tv_sec, tvd.tv_usec);
+        hand(0);	// effectively an exit
+    }
+    if (tcpport!=0) {
+    	wait_for_client();
+    }
+
+    start_time = time(NULL);
+
+	int measurement_interval_pos = -1; // extract measurementInterval if present
+	char *field_names[numberoffields];
+    for(y=0; y<numberoffields;y++) {
+		field_names[y] = strdup(ftaschema_field_name(schema,y));
+		if(strcmp(field_names[y], "measurementInterval")==0)
+			measurement_interval_pos = y;
+	}
+
+
+	struct timeval tsample;
+	gettimeofday(&tsample, 0);
+	char start_ts[100], curr_ts[100];
+	sprintf(start_ts,"%ld%06d", tsample.tv_sec, tsample.tv_usec);
+
+	long unsigned int lineno=0;
+	long unsigned int seqno=0;
+	unsigned int measurement_interval;
+    while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {
+		lineno++;
+        if (dump)	// -D in command line
+            continue;
+        if (ftaschema_is_eof_tuple(schema, rbuf)) {
+            /* initiate shutdown or something of that nature */
+            printf("#All data proccessed\n");
+            exit(0);
+        }
+        if (!rsize)
+            continue;
+        if (verbose >=2) {
+            snprintf(linebuf,MAXLINE,"RESULT CODE => %u\n",code);
+            emit_line();
+        }
+        if ((code==0)&&(rfta_id.streamid == fta_id.streamid)) {
+			seqno++;
+			gettimeofday(&tsample, 0);
+			sprintf(curr_ts,"%ld%06d", tsample.tv_sec, tsample.tv_usec);
+			int pos;
+			if(ves_version < 7){
+				pos = snprintf(linebuf, MAXLINE,
+  "{\"event\": { \"commonEventHeader\": { "
+        "\"domain\": \"measurementsForVfScaling\", "
+        "\"eventId\": \"%s%u\", "
+        "\"eventType\": \"%s\", "
+        "\"eventName\": \"Measurement_MC_%s\", "
+        "\"lastEpochMicrosec\": %s, "
+        "\"priority\": \"Normal\", "
+        "\"reportingEntityName\": \"GS-LITE MC\", "
+        "\"sequence\": %u, "
+        "\"sourceName\": \"meas_cmpgn_xapp\", "
+        "\"startEpochMicrosec\": %s, "
+        "\"version\": 5 "
+      "}, "
+      "\"measurementsForVfScalingFields\": { "
+	  	"\"additionalFields\": ["
+				,argv[0],lineno, argv[0], argv[0], curr_ts, seqno, start_ts
+				);
+			}else{
+				pos = snprintf(linebuf, MAXLINE,
+  "{\"event\": { \"commonEventHeader\": { "
+        "\"domain\": \"measurement\", "
+        "\"eventId\": \"%s%u\", "
+        "\"eventType\": \"%s\", "
+        "\"eventName\": \"Measurement_MC_%s\", "
+        "\"lastEpochMicrosec\": %s, "
+        "\"priority\": \"Normal\", "
+        "\"reportingEntityName\": \"GS-LITE MC\", "
+        "\"sequence\": %u, "
+        "\"sourceName\": \"meas_cmpgn_xapp\", "
+        "\"startEpochMicrosec\": %s, "
+        "\"version\": \"4.0.1\", "
+        "\"vesEventListenerVersion\": \"7.0.1\" "
+      "}, "
+      "\"measurementFields\": { "
+	  	"\"additionalFields\": {"
+				,argv[0],lineno, argv[0], argv[0], curr_ts, seqno, start_ts
+				);
+			}
+
+			measurement_interval = 0;
+            for(y=0; y<numberoffields;y++) {
+                struct access_result ar;
+//                if (verbose>=2)
+//                    printf("%s->",ftaschema_field_name(schema,y));
+              	if(y>0){
+					linebuf[pos]=',';
+					pos++;
+				}
+                ar=ftaschema_get_field_by_index(schema,y,rbuf,rsize);
+                switch (ar.field_data_type) {
+                    case INT_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%d\"}",field_names[y], ar.r.i);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%d\"",field_names[y], ar.r.i);
+						if(y==measurement_interval_pos)
+							measurement_interval = (unsigned int)ar.r.i;
+                        break;
+                    case UINT_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u\"}",field_names[y], ar.r.ui);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%u\"",field_names[y], ar.r.ui);
+						if(y==measurement_interval_pos)
+							measurement_interval = (unsigned int)ar.r.ui;
+                        break;
+                    case IP_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u.%u.%u.%u\"}",field_names[y], ar.r.ui>>24&0xff,
+                                 ar.r.ui>>16&0xff,
+                                 ar.r.ui>>8&0xff,
+                                 ar.r.ui&0xff);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%u.%u.%u.%u\"",field_names[y], ar.r.ui>>24&0xff,
+                                 ar.r.ui>>16&0xff,
+                                 ar.r.ui>>8&0xff,
+                                 ar.r.ui&0xff);
+                        break;
+                    case IPV6_TYPE:
+                    {
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"",field_names[y]);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"",field_names[y]);
+                        unsigned x;
+                        unsigned zc=0;
+                        for(x=0;x<4;x++) { if (ar.r.ip6.v[x]==0) zc++;}
+                        if (zc!=4) {
+                            snprintf(linebuf,MAXLINE,"");
+                            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]);
+                                pos += snprintf(linebuf+pos,MAXLINE-pos,"%04x",y);
+                                if (x<7){
+									pos += snprintf(linebuf+pos,MAXLINE-pos,":");
+								}
+                            }
+                        } else {
+                            pos+=snprintf(linebuf+pos,MAXLINE-pos,"::");
+                        }
+						if(ves_version < 7)
+							pos += snprintf(linebuf+pos, MAXLINE-pos,"\"}");
+						else
+							pos += snprintf(linebuf+pos, MAXLINE-pos,"\"");
+                    }
+                        break;
+                        
+                    case USHORT_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%u\"}",field_names[y], ar.r.ui);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\":  \"%u\"}",field_names[y], ar.r.ui);
+						if(y==measurement_interval_pos)
+							measurement_interval = (unsigned int)ar.r.ui;
+                        break;
+                    case BOOL_TYPE:
+						if(ves_version < 7){
+	                        if (ar.r.ui==0) {
+   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"FALSE\"}",field_names[y]);
+   		                     } else {
+   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"TRUE\"}",field_names[y]);
+   	                     	}
+						}else{
+	                        if (ar.r.ui==0) {
+   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"FALSE\"",field_names[y]);
+   		                     } else {
+   		                     	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"TRUE\"",field_names[y]);
+   	                     	}
+						}
+                        break;
+                    case ULLONG_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%llu\"}",field_names[y], ar.r.ul);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%llu\"",field_names[y], ar.r.ul);
+						if(y==measurement_interval_pos)
+							measurement_interval = (unsigned int)ar.r.ul;
+                        break;
+                    case LLONG_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%lld\"}",field_names[y], ar.r.l);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%lld\"",field_names[y], ar.r.l);
+						if(y==measurement_interval_pos)
+							measurement_interval = (unsigned int)ar.r.l;
+                        break;
+                    case FLOAT_TYPE:
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%f\"}",field_names[y], ar.r.f);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%f\"",field_names[y], 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;
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%f\"}",field_names[y], t);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%f\"",field_names[y], t);
+                    }
+                        break;
+                    case VSTR_TYPE:
+                    {
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"",field_names[y]);
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"",field_names[y]);
+                        int x;
+                        int c;
+                        char * src;
+                        src=(char*)ar.r.vs.offset;
+						for(x=0;x<ar.r.vs.length;x++){
+                            c=src[x];
+                            if ((c<='~') && (c>=' ')) {
+                                if (pos<MAXLINE-1) {
+                                    linebuf[pos]=c;
+                                    pos++;
+                                }
+                            } else {
+                                if (pos<MAXLINE-1) {
+                                    linebuf[pos]='.';
+                                    pos++;
+                                }
+                            }
+						}
+						if(ves_version < 7)
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"}");
+						else
+                        	pos += snprintf(linebuf+pos,MAXLINE-pos,"\"");
+                    }
+                        break;
+                    default:
+                        linebuf[0]=0;
+                        break;
+                }
+			}
+			if(ves_version < 7){
+				snprintf(linebuf+pos, MAXLINE-pos,
+	  	"], \"measurementInterval\": %u, \"measurementsForVfScalingVersion\": 1"
+		"}}}\n", measurement_interval
+				);
+			}else{
+				snprintf(linebuf+pos, MAXLINE-pos,
+	  	"}, \"measurementInterval\": %u, \"measurementFieldsVersion\": \"4.0\""
+		"}}}\n", measurement_interval
+				);
+			}
+			if(curl_address==NULL){
+            	emit_line();
+			}else{
+				http_post_request_hdr(curl_endpoint, curl_url, linebuf, &http_code, curl_auth);
+				if(http_code != 200 && http_code != 202){
+					gslog(LOG_WARNING, "http return code is %d\n",http_code);
+				}
+			}
+            if (verbose!=0) fflush(stdout);
+        } else {
+            if (rfta_id.streamid != fta_id.streamid)
+                fprintf(stderr,"Got unkown streamid %llu \n",rfta_id.streamid);
+        }
+
+        // whenever we receive a temp tuple check if we reached time limit
+        if ((code==2)  && tlimit && (time(NULL)-start_time)>=tlimit) {
+            fprintf(stderr,"Reached time limit of %d seconds\n",tlimit);    
+            ftaapp_exit();
+            exit(0);
+        }        
+    }
+}
+