/* ------------------------------------------------
 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 <errno.h>
#include <string.h>

#ifdef RMR_ENABLED
#include <rmr/rmr.h>
#include <rmr/RIC_message_types.h>
#include <sys/epoll.h>
#endif

#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;

// how frequently we will log stats (expressed in tuples posted)
#define STAT_FREQUENCY 5


// 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");
            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");
            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	rmr_port = NULL;    

#ifdef RMR_ENABLED
    // RMR-related parameters
    gs_int32_t	rmr_mtype = MC_REPORT;
#endif

	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;

    gs_uint64_t post_success_cnt = 0ULL;
    gs_uint64_t post_failure_cnt = 0ULL;    

#ifdef RMR_ENABLED
	void* mrc;      						//msg router context
	struct epoll_event events[1];			// list of events to give to epoll
	struct epoll_event epe;                 // event definition for event to listen to
	gs_int32_t     ep_fd = -1;						// epoll's file des (given to epoll_wait)
	gs_int32_t rcv_fd;     						// file des that NNG tickles -- give this to epoll to listen on
	gs_int32_t nready;								// number of events ready for receive
	rmr_mbuf_t*		rmr_sbuf;					// send buffer
	rmr_mbuf_t*		rmr_rbuf;					// received buffer

    gs_uint64_t rmr_post_success_cnt = 0ULL;
    gs_uint64_t rmr_post_failure_cnt = 0ULL;        
#endif
    
	gsopenlog(argv[0]);
    
    while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:R: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 'R':
                rmr_port=strdup(optarg);
                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>] [-R <rmr_port>] <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");
    }


//	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;

    if (rmr_port) {
#ifdef RMR_ENABLED        
        /* initialize RMR library */
        if( (mrc = rmr_init( rmr_port, 1400, RMRFL_NONE )) == NULL ) {
            fprintf(stderr, "%s::error:unable to initialise RMR\n", me);
            exit( 1 );
        }

        rcv_fd = rmr_get_rcvfd( mrc );					// set up epoll things, start by getting the FD from MRr
        if( rcv_fd < 0 ) {
            fprintf(stderr, "%s::error:unable to set up polling fd\n", me);
            exit( 1 );
        }
        if( (ep_fd = epoll_create1( 0 )) < 0 ) {
            fprintf(stderr, "%s::error:unable to create epoll fd: %d\n", me, errno);
            exit( 1 );
        }
        epe.events = EPOLLIN;
        epe.data.fd = rcv_fd;

        if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
            fprintf(stderr, "%s::error:epoll_ctl status not 0 : %s\n", me, strerror(errno));
            exit( 1 );
        }

        rmr_sbuf = rmr_alloc_msg( mrc, MAXLINE );	// alloc first send buffer; subsequent buffers allcoated on send
        rmr_rbuf = NULL;						// don't need to alloc receive buffer

        while( ! rmr_ready( mrc ) ) {		// must have a route table before we can send; wait til RMr say it has one
            sleep( 10 );
        }
        fprintf( stderr, "%s: RMR is ready\n", argv[0]);       
#else
		fprintf(stderr,"Runtime libraries built without RMR support. Rebuild with RMR_ENABLED defined in gsoptions.h\n");
		exit(0);
#endif
    }
        
    /* 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;
	double 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 = (double)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 = (double)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 = (double)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 = (double)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 = (double)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);
						if(y==measurement_interval_pos)
							measurement_interval = (double)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\": %f, \"measurementsForVfScalingVersion\": 1"
		"}}}\n", measurement_interval
				);
			}else{
				snprintf(linebuf+pos, MAXLINE-pos,
	  	"}, \"measurementInterval\": %f, \"measurementFieldsVersion\": \"4.0\""
		"}}}\n", measurement_interval
				);
			}

#ifdef RMR_ENABLED
            if (rmr_port) {                
                rmr_sbuf->mtype = rmr_mtype;							// fill in the message bits
		        rmr_sbuf->len =  strlen(linebuf) + 1;           // our receiver likely wants a nice acsii-z string
		        memcpy(rmr_sbuf->payload, linebuf, rmr_sbuf->len);
                rmr_sbuf->state = 0;
                rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf);				// send it (send returns an empty payload on success, or the original payload on fail/retry)
                while( rmr_sbuf->state == RMR_ERR_RETRY ) {			// soft failure (device busy?) retry
                    rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf);			// retry send until it's good (simple test; real programmes should do better)
                }
                if(rmr_sbuf->state != RMR_OK) {
                    gslog(LOG_WARNING, "rmr_send_msg() failure, strerror(errno) is %s", strerror(errno));
                    rmr_post_failure_cnt++;
                } else
                    rmr_post_success_cnt++;
                if (((rmr_post_success_cnt+rmr_post_failure_cnt) % STAT_FREQUENCY) == 0)
                    gslog(LOG_WARNING, "%s: successful RMR posts - %llu, failed RMR posts - %llu", argv[0], rmr_post_success_cnt, rmr_post_failure_cnt);
            }
#endif            

			if(curl_address==NULL){
                if (!rmr_port)  // if neither VES collector nor RMR is specified print to standard output
            	    emit_line();
			}else{
				http_post_request_hdr(curl_endpoint, curl_url, linebuf, &http_code, curl_auth);
				if(http_code != 200 && http_code != 202){
                    post_failure_cnt++; 
					gslog(LOG_WARNING, "http return code is %d",http_code);
				} else {
                    post_success_cnt++;   
                }  
                if (((post_success_cnt+post_failure_cnt) % STAT_FREQUENCY) == 0)
                    gslog(LOG_WARNING, "%s: successful ves posts - %llu, failed ves posts - %llu", argv[0], post_success_cnt, post_failure_cnt);
			}
            if (verbose!=0) fflush(stdout);
        } else {
            if (rfta_id.streamid != fta_id.streamid)
                fprintf(stderr,"Got unknown 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);
        }        
    }
}