INFO: The new Git default branch name is "main". Details here: https://about.gitlab.com/blog/2021/03/10/new-git-default-branch-name/

nmxp.c 32.3 KB
Newer Older
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
1
2
/*! \file
 *
Matteo Quintiliani's avatar
Matteo Quintiliani committed
3
 * \brief Nanometrics Protocol Library
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
4
5
6
7
8
9
 *
 * Author:
 * 	Matteo Quintiliani
 * 	Istituto Nazionale di Geofisica e Vulcanologia - Italy
 *	quintiliani@ingv.it
 *
10
 * $Id: nmxp.c,v 1.74 2008-03-18 15:19:37 mtheo Exp $
Matteo Quintiliani's avatar
Matteo Quintiliani committed
11
 *
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
12
13
14
15
 */

#include "nmxp.h"

Matteo Quintiliani's avatar
Matteo Quintiliani committed
16
17
#include "config.h"

Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
18
#include <stdio.h>
19
#include <stdlib.h>
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
20
21
#include <string.h>

22
23
24
25
#ifdef HAVE_GETTIMEOFDAY
#include <sys/time.h>
#endif

Matteo Quintiliani's avatar
Matteo Quintiliani committed
26
27
28
29
#ifdef HAVE_ARPA_INET_H
#include <arpa/inet.h>
#endif

30
31
32
33
#ifdef HAVE_WINDOWS_H
#include "winsock2.h"
#endif

34
int nmxp_sendConnect(int isock) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
35
    return nmxp_sendMessage(isock, NMXP_MSG_CONNECT, NULL, 0);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
36
37
}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
38
39
int nmxp_sendTerminateSubscription(int isock, NMXP_SHUTDOWN_REASON reason, char *message) {
    return nmxp_sendMessage(isock, NMXP_MSG_TERMINATESUBSCRIPTION, message, ((message)? strlen(message)-1 : 0));
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
40
41
}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
42
int nmxp_receiveChannelList(int isock, NMXP_CHAN_LIST **pchannelList) {
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
43
44
    int ret;
    int i;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
45
    int recv_errno;
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
46

Matteo Quintiliani's avatar
Matteo Quintiliani committed
47
    NMXP_MSG_SERVER type;
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
48
    void *buffer;
49
    int32_t length;
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
50
51
52

    *pchannelList = NULL;

Matteo Quintiliani's avatar
Matteo Quintiliani committed
53
    ret = nmxp_receiveMessage(isock, &type, &buffer, &length, 0, &recv_errno);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
54

Matteo Quintiliani's avatar
Matteo Quintiliani committed
55
    if(type != NMXP_MSG_CHANNELLIST) {
56
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_PACKETMAN, "Type %d is not NMXP_MSG_CHANNELLIST!\n", type);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
57
58
59
60
61
    } else {

	*pchannelList = buffer;
	(*pchannelList)->number = ntohl((*pchannelList)->number);

62
	nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CHANNEL, "number of channels %d\n", (*pchannelList)->number);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
63
	
64
	/* TODO check*/
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
65
66
67

	for(i=0; i < (*pchannelList)->number; i++) {
	    (*pchannelList)->channel[i].key = ntohl((*pchannelList)->channel[i].key);
68
69
70
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CHANNEL, "%12d %s\n",
		    (*pchannelList)->channel[i].key,
		    NMXP_LOG_STR((*pchannelList)->channel[i].name));
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
71
72
73
74
75
76
77
78
	}

    }

    return ret;
}


79
int nmxp_sendAddTimeSeriesChannel_raw(int isock, NMXP_CHAN_LIST_NET *channelList, int32_t shortTermCompletion, int32_t out_format, NMXP_BUFFER_FLAG buffer_flag) {
80
    int ret;
81
    int32_t buffer_length = 16 + (4 * channelList->number); 
82
    char *buffer = malloc(buffer_length);
83
    int32_t app, i, disp;
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

    disp=0;

    app = htonl(channelList->number);
    memcpy(&buffer[disp], &app, 4);
    disp+=4;

    for(i=0; i < channelList->number; i++) {
	app = htonl(channelList->channel[i].key);
	memcpy(&buffer[disp], &app, 4);
	disp+=4;
    }
    
    app = htonl(shortTermCompletion);
    memcpy(&buffer[disp], &app, 4);
    disp+=4;

    app = htonl(out_format);
    memcpy(&buffer[disp], &app, 4);
    disp+=4;

    app = htonl(buffer_flag);
    memcpy(&buffer[disp], &app, 4);
    disp+=4;

Matteo Quintiliani's avatar
Matteo Quintiliani committed
109
    ret = nmxp_sendMessage(isock, NMXP_MSG_ADDTIMESERIESCHANNELS, buffer, buffer_length);
110
111
112
113
114
115
116

    if(buffer) {
	free(buffer);
    }
    return ret;
}

117
int nmxp_sendAddTimeSeriesChannel(int isock, NMXP_CHAN_LIST_NET *channelList, int32_t shortTermCompletion, int32_t out_format, NMXP_BUFFER_FLAG buffer_flag, const int n_channel, const int n_usec) {
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
    static int i = 0;
    static int first_time = 1;
    static struct timeval last_tp_now;

    int j;
    int ret = 0;
    NMXP_CHAN_LIST_NET split_channelList;
    long diff_usec;
    struct timeval tp_now;

#ifdef HAVE_GETTIMEOFDAY
    gettimeofday(&tp_now, NULL);
#else
    TODO gettimeofday not found
#endif

    if(i <  channelList->number) {
	    if(first_time) {
		    diff_usec = n_usec + 1;
		    first_time = 0;
		    last_tp_now.tv_sec = 0;
		    last_tp_now.tv_usec = 0;
	    } else {
		    diff_usec = (tp_now.tv_sec - last_tp_now.tv_sec) * 1000000;
		    diff_usec += (tp_now.tv_usec - last_tp_now.tv_usec);
	    }
	    if(diff_usec >= n_usec) {
		    /* while(ret == 0  &&  i <  channelList->number) { */
		    split_channelList.number = 0;
		    while(split_channelList.number < n_channel  &&  i < channelList->number) {
			    split_channelList.channel[split_channelList.number].key = channelList->channel[i].key;
			    /* Not necessary, but it could help for debugging */
			    strcpy(split_channelList.channel[split_channelList.number].name, channelList->channel[i].name);
			    split_channelList.number++;
			    i++;
		    }
		    if(split_channelList.number > 0) {
			    nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_ANY,
					    "Added %d channels (%d):", split_channelList.number, diff_usec);
			    for(j=0; j < split_channelList.number; j++) {
				    nmxp_log(NMXP_LOG_NORM_NO, NMXP_LOG_D_ANY, " %s", NMXP_LOG_STR(split_channelList.channel[j].name));
			    }
			    nmxp_log(NMXP_LOG_NORM_NO, NMXP_LOG_D_ANY, "\n");
			    ret = nmxp_sendAddTimeSeriesChannel_raw(isock, &split_channelList, shortTermCompletion, out_format, buffer_flag);
		    }
		    /* } */
		    last_tp_now.tv_sec = tp_now.tv_sec;
		    last_tp_now.tv_usec = tp_now.tv_usec;
	    }
    }

    return ret;
}

172

Matteo Quintiliani's avatar
Matteo Quintiliani committed
173
NMXP_DATA_PROCESS *nmxp_receiveData(int isock, NMXP_CHAN_LIST_NET *channelList, const char *network_code, int timeoutsec, int *recv_errno ) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
174
    NMXP_MSG_SERVER type;
175
    void *buffer = NULL;
176
    int32_t length;
177
    NMXP_DATA_PROCESS *pd = NULL;
178

Matteo Quintiliani's avatar
Matteo Quintiliani committed
179
    if(nmxp_receiveMessage(isock, &type, &buffer, &length, timeoutsec, recv_errno) == NMXP_SOCKET_OK) {
180
	if(type == NMXP_MSG_COMPRESSED) {
181
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_PACKETMAN, "Type %d is NMXP_MSG_COMPRESSED!\n", type);
182
	    pd = nmxp_processCompressedData(buffer, length, channelList, network_code);
183
	} else if(type == NMXP_MSG_DECOMPRESSED) {
184
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_PACKETMAN, "Type %d is NMXP_MSG_DECOMPRESSED!\n", type);
185
	    pd = nmxp_processDecompressedData(buffer, length, channelList, network_code);
186
	} else {
187
	    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_PACKETMAN, "Type %d is not NMXP_MSG_COMPRESSED or NMXP_MSG_DECOMPRESSED!\n", type);
188
	}
Matteo Quintiliani's avatar
Matteo Quintiliani committed
189
190
    }

191
    return pd;
192
193
}

194

195
int nmxp_sendConnectRequest(int isock, char *naqs_username, char *naqs_password, int32_t connection_time) {
196
    int ret;
197
    int i;
198
    char crc32buf[100];
199
200
    char *pcrc32buf = crc32buf;
    int crc32buf_length = 0;
201
    NMXP_CONNECT_REQUEST connectRequest;
202
    int naqs_username_length, naqs_password_length;
203
    int32_t protocol_version = 0;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
204
    char *pp = NULL;
205

206
207
    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CRC, "%s - %s\n",
	    NMXP_LOG_STR(naqs_username), NMXP_LOG_STR(naqs_password));
208

209
210
211
    naqs_username_length = (naqs_username)? strlen(naqs_username) : 0;
    naqs_password_length = (naqs_password)? strlen(naqs_password) : 0;

212
213
214
215
216
217
218
219
    for(i=0; i < 100; i++) {
	crc32buf[i] = 0;
    }

    for(i=0; i<12; i++) {
	connectRequest.username[i] = 0;
    }
    if(naqs_username_length != 0) {
220
	strcpy(connectRequest.username, naqs_username);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
221
    }
222
223
224
225

    connectRequest.version = protocol_version;
    connectRequest.connection_time = connection_time;

226
227
228
    if(!nmxp_data_bigendianhost()) {
	nmxp_data_swap_4b ((int32_t *)&connection_time);
    }
229

230
    if(naqs_username_length == 0  &&  naqs_password_length == 0 ) {
231
232
	/* sprintf(crc32buf, "%d%d", protocol_version, connection_time); */
	/* TODO */
233
    } else if(naqs_username_length != 0  &&  naqs_password_length != 0 ) {
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
	/* sprintf(crc32buf, "%s%d%d%s", naqs_username, protocol_version,
		connection_time, naqs_password); */

	memcpy(pcrc32buf, naqs_username, naqs_username_length);
	crc32buf_length += naqs_username_length;
	pcrc32buf = crc32buf + crc32buf_length;

	memcpy(pcrc32buf, &(protocol_version), sizeof(protocol_version));
	crc32buf_length += sizeof(protocol_version);
	pcrc32buf = crc32buf + crc32buf_length;

	memcpy(pcrc32buf, &(connection_time), sizeof(connection_time));
	crc32buf_length += sizeof(connection_time);
	pcrc32buf = crc32buf + crc32buf_length;

	memcpy(pcrc32buf, naqs_password, naqs_password_length);
	crc32buf_length += naqs_password_length;
	pcrc32buf = crc32buf + crc32buf_length;

253
    } else if(naqs_username_length != 0 ) {
254
255
	/* sprintf(crc32buf, "%s%d%d", naqs_username, protocol_version, connection_time); */
	/* TODO */
256
    } else if(naqs_password_length != 0 ) {
257
258
	/* sprintf(crc32buf, "%d%d%s", protocol_version, connection_time, naqs_password); */
	/* TODO */
259
    }
260
261
262
    connectRequest.version = htonl(connectRequest.version);
    connectRequest.connection_time = htonl(connectRequest.connection_time);
    connectRequest.crc32 = htonl(crc32(0L, crc32buf, crc32buf_length));
263
264
265
266

    ret = nmxp_sendMessage(isock, NMXP_MSG_CONNECTREQUEST, &connectRequest, sizeof(NMXP_CONNECT_REQUEST));

    if(ret == NMXP_SOCKET_OK) {
267
268
	nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CRC, "Send a ConnectRequest crc32buf length %d, crc32 = %d\n",
		crc32buf_length, connectRequest.crc32);
269
270
271
	for(i=0; i < crc32buf_length; i++) {
	    nmxp_log(NMXP_LOG_NORM_NO, NMXP_LOG_D_CRC, "%d ", crc32buf[i]);
	}
Matteo Quintiliani's avatar
Matteo Quintiliani committed
272
	pp = (char *) &connectRequest.crc32;
273
274
275
276
	for(i=0; i < sizeof(connectRequest.crc32); i++) {
	    nmxp_log(NMXP_LOG_NORM_NO, NMXP_LOG_D_CRC, "%d ", pp[i]);
	}
	nmxp_log(NMXP_LOG_NORM_NO, NMXP_LOG_D_CRC, "\n");
277
    } else {
278
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CRC, "Send a ConnectRequest.\n");
279
280
281
282
283
284
    }

    return ret;
}


285
int nmxp_readConnectionTime(int isock, int32_t *connection_time) {
286
    int ret;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
287
288
    int recv_errno;
    ret = nmxp_recv_ctrl(isock, connection_time, sizeof(int32_t), 0, &recv_errno);
289
    *connection_time = ntohl(*connection_time);
290
    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CONNFLOW, "Read connection time from socket %d.\n", *connection_time);
291
    if(ret != NMXP_SOCKET_OK) {
292
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Read connection time from socket.\n");
293
294
295
296
297
    }
    return ret;
}


298
int nmxp_waitReady(int isock) {
299
300
    int times = 0;
    int rc = NMXP_SOCKET_OK;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
301
302
303
    int32_t signature;
    int32_t type = 0;
    int32_t length;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
304
    int recv_errno;
305
306

    while(rc == NMXP_SOCKET_OK  &&  type != NMXP_MSG_READY) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
307
	rc = nmxp_recv_ctrl(isock, &signature, sizeof(signature), 0, &recv_errno);
308
309
310
	if(rc != NMXP_SOCKET_OK) return rc;
	signature = ntohl(signature);
	if(signature == 0) {
311
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CONNFLOW, "signature is equal to zero. receive again.\n");
Matteo Quintiliani's avatar
Matteo Quintiliani committed
312
	    rc = nmxp_recv_ctrl(isock, &signature, sizeof(signature), 0, &recv_errno);
313
314
315
	    signature = ntohl(signature);
	}
	if(signature != NMX_SIGNATURE) {
316
	    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "signature is not valid. signature = %d\n", signature);
317
318
319
320
	    if(signature == 200) {
		    int32_t err_length;
		    int32_t err_reason;
		    char err_buff[200];
Matteo Quintiliani's avatar
Matteo Quintiliani committed
321
		    rc = nmxp_recv_ctrl(isock, &err_length, sizeof(err_length), 0, &recv_errno);
322
		    err_length = ntohl(err_length);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
323
		    rc = nmxp_recv_ctrl(isock, &err_reason, sizeof(err_reason), 0, &recv_errno);
324
325
		    err_reason = ntohl(err_reason);
		    if(err_length > 4) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
326
			    rc = nmxp_recv_ctrl(isock, err_buff, err_length-4, 0, &recv_errno);
327
328
			    err_buff[err_length] = 0;
		    }
329
		    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "TerminateMessage from Server: %s (%d).\n",
330
			    NMXP_LOG_STR(err_buff), err_reason);
331
	    }
332
333
334
	    return NMXP_SOCKET_ERROR;
	}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
335
	rc = nmxp_recv_ctrl(isock, &type, sizeof(type), 0, &recv_errno);
336
337
338
	if(rc != NMXP_SOCKET_OK) return rc;
	type = ntohl(type);
	if(type != NMXP_MSG_READY) {
339
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CONNFLOW, "type is not READY. type = %d\n", type);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
340
	    rc = nmxp_recv_ctrl(isock, &length, sizeof(length), 0, &recv_errno);
341
342
343
	    if(rc != NMXP_SOCKET_OK) return rc;
	    length = ntohl(length);
	    if(length > 0) {
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
		if(type == NMXP_MSG_TERMINATESUBSCRIPTION) {
		    char *str_msg = NULL;
		    char *buf_app = (char *) malloc(sizeof(char) * length);
		    int32_t reason;
		    memcpy(&reason, buf_app, sizeof(reason));
		    reason = ntohl(reason);
		    str_msg = buf_app + sizeof(reason);
		    rc = nmxp_recv_ctrl(isock, buf_app, length, 0, &recv_errno);
		    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_ANY, "%d %s shutdown: %s\n",
			    reason,
			    (reason == 0)? "Normal" : (reason == 1)? "Error" : (reason == 2)? "Timeout" : "Unknown",
			    str_msg);
		    if(buf_app) {
			free(buf_app);
		    }
359
360
361
		    /* Close the socket*/
		    nmxp_closeSocket(isock);
		    exit(-1);
362
		} else if(length == 4) {
363
		    int32_t app;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
364
		    rc = nmxp_recv_ctrl(isock, &app, length, 0, &recv_errno);
365
366
		    if(rc != NMXP_SOCKET_OK) return rc;
		    app = ntohl(app);
367
		    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CONNFLOW, "value = %d\n", app);
368
369
		} else {
		    char *buf_app = (char *) malloc(sizeof(char) * length);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
370
		    rc = nmxp_recv_ctrl(isock, buf_app, length, 0, &recv_errno);
371
372
373
374
375
376
		    if(buf_app) {
			free(buf_app);
		    }
		}
	    }
	} else {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
377
	    rc = nmxp_recv_ctrl(isock, &length, sizeof(length), 0, &recv_errno);
378
379
380
	    if(rc != NMXP_SOCKET_OK) return rc;
	    length = ntohl(length);
	    if(length != 0) {
381
		nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "length is not equal to zero. length = %d\n", length);
382
383
384
385
386
387
		return NMXP_SOCKET_ERROR;
	    }
	}

	times++;
	if(times > 10) {
388
	    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "waiting_ready_message. times > 10\n");
389
390
391
392
393
394
395
396
397
	    rc = NMXP_SOCKET_ERROR;
	}

    }

    return rc;
}


398
int nmxp_sendDataRequest(int isock, int32_t key, int32_t start_time, int32_t end_time) {
399
400
401
402
403
404
405
406
407
408
    int ret;
    NMXP_DATA_REQUEST dataRequest;

    dataRequest.chan_key = htonl(key);
    dataRequest.start_time = htonl(start_time);
    dataRequest.end_time = htonl(end_time);

    ret = nmxp_sendMessage(isock, NMXP_MSG_DATAREQUEST, &dataRequest, sizeof(dataRequest));

    if(ret != NMXP_SOCKET_OK) {
409
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Send a Request message\n");
410
411
412
413
414
415
    }

    return ret;
}


416
417
418
NMXP_CHAN_LIST *nmxp_getAvailableChannelList(char * hostname, int portnum, NMXP_DATATYPE datatype) {
    int naqssock;
    NMXP_CHAN_LIST *channelList = NULL, *channelList_subset = NULL;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
419
    /* int i; */
420

421
    /* 1. Open a socket*/
422
423
424
425
    naqssock = nmxp_openSocket(hostname, portnum);

    if(naqssock != NMXP_SOCKET_ERROR) {

426
	/* 2. Send a Connect*/
427
428
	if(nmxp_sendConnect(naqssock) == NMXP_SOCKET_OK) {

429
	    /* 3. Receive ChannelList*/
430
431
432
	     if(nmxp_receiveChannelList(naqssock, &channelList) == NMXP_SOCKET_OK) {

		 channelList_subset = nmxp_chan_getType(channelList, datatype);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
433
		 nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CHANNEL, "%d / %d are DataType channel.\n", channelList_subset->number, channelList->number);
434

435
		 /* nmxp_chan_sortByKey(channelList_subset);*/
436
437
		 nmxp_chan_sortByName(channelList_subset);

Matteo Quintiliani's avatar
Matteo Quintiliani committed
438
		 /*
439
		 for(i=0; i < channelList_subset->number; i++) {
440
441
442
		     nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_CHANNEL, "%12d %s\n",
			     channelList_subset->channel[i].key,
			     NMXP_LOG_STR(channelList_subset->channel[i].name));
443
		 }
Matteo Quintiliani's avatar
Matteo Quintiliani committed
444
		 */
445

446
		 /* 4. Send a Request Pending (optional)*/
447

448
		 /* 5. Send AddChannels*/
449

450
		 /* 6. Repeat until finished: receive and handle packets*/
451

452
		 /* 7. Send Terminate Subscription*/
453
454
455
		 nmxp_sendTerminateSubscription(naqssock, NMXP_SHUTDOWN_NORMAL, "Good Bye!");

	     } else {
456
		 nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CHANNEL, "Error on receiveChannelList()\n");
457
458
	     }
	} else {
459
	    nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CHANNEL, "Error on sendConnect()\n");
460
461
	}

462
	/* 8. Close the socket*/
463
464
465
466
467
468
469
470
471
472
	nmxp_closeSocket(naqssock);
    }

    if(channelList) {
	free(channelList);
    }

    return channelList_subset;
}

473

474
NMXP_META_CHAN_LIST *nmxp_getMetaChannelList(char * hostname, int portnum, NMXP_DATATYPE datatype, int flag_request_channelinfo, char *datas_username, char *datas_password, NMXP_CHAN_LIST **pchannelList) {
475
476
    int naqssock;
    NMXP_CHAN_PRECISLIST *precisChannelList = NULL;
477
478
479
480
    NMXP_CHAN_LIST *channelList = NULL;
    NMXP_META_CHAN_LIST *chan_list = NULL;
    NMXP_META_CHAN_LIST *iter = NULL;
    int i = 0;
481
    int32_t connection_time;
482
    int ret_sock;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
483
    int recv_errno;
484
    
485
486
    NMXP_MSG_SERVER type;
    void *buffer = NULL;
487
    int32_t length;
488
489
490
    NMXP_PRECISLISTREQUEST precisListRequestBody;
    NMXP_CHANNELINFOREQUEST channelInfoRequestBody;
    NMXP_CHANNELINFORESPONSE *channelInfo;
491
492
493
494
495
496
497

    char str_start[200], str_end[200];
    str_start[0] = 0;
    str_end[0] = 0;
    
    /* DAP Step 1: Open a socket */
    if( (naqssock = nmxp_openSocket(hostname, portnum)) == NMXP_SOCKET_ERROR) {
498
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Error opening socket!\n");
499
500
501
502
503
	return NULL;
    }

    /* DAP Step 2: Read connection time */
    if(nmxp_readConnectionTime(naqssock, &connection_time) != NMXP_SOCKET_OK) {
504
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Error reading connection time from server!\n");
505
506
507
508
509
	return NULL;
    }

    /* DAP Step 3: Send a ConnectRequest */
    if(nmxp_sendConnectRequest(naqssock, datas_username, datas_password, connection_time) != NMXP_SOCKET_OK) {
510
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Error sending connect request!\n");
511
512
513
514
515
	return NULL;
    }

    /* DAP Step 4: Wait for a Ready message */
    if(nmxp_waitReady(naqssock) != NMXP_SOCKET_OK) {
516
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CONNFLOW, "Error waiting Ready message!\n");
517
518
519
	return NULL;
    }

520
521
522
523
524
525



    /* DAP Step 5: Send Data Request */
    nmxp_sendHeader(naqssock, NMXP_MSG_CHANNELLISTREQUEST, 0);
    /* DAP Step 6: Receive Data until receiving a Ready message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
526
    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
527
    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
528
529
530
531
532
533
534
535

    while(ret_sock == NMXP_SOCKET_OK   &&    type != NMXP_MSG_READY) {
	channelList = buffer;

	channelList->number = ntohl(channelList->number);

	for(i = 0; i < channelList->number; i++) {
	    channelList->channel[i].key = ntohl(channelList->channel[i].key);
536
	    if(getDataTypeFromKey(channelList->channel[i].key) == datatype) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
537
		nmxp_meta_chan_add(&chan_list, channelList->channel[i].key, channelList->channel[i].name, 0, 0, NULL, NMXP_META_SORT_NAME);
538
	    }
539
540
541
	}

	/* Receive Message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
542
	ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
543
	nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
544
545
    }

546
    *pchannelList = channelList;
547

548
549
550
551
    /* DAP Step 5: Send Data Request */
    precisListRequestBody.instr_id = htonl(-1);
    precisListRequestBody.datatype = htonl(NMXP_DATA_TIMESERIES);
    precisListRequestBody.type_of_channel = htonl(-1);
552
    nmxp_sendMessage(naqssock, NMXP_MSG_PRECISLISTREQUEST, &precisListRequestBody, sizeof(NMXP_PRECISLISTREQUEST));
553

554
555

    /* DAP Step 6: Receive Data until receiving a Ready message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
556
    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
557
    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
558
559
560
561
562
563
564
565
566
567
568
569
570

    while(ret_sock == NMXP_SOCKET_OK   &&    type != NMXP_MSG_READY) {
	precisChannelList = buffer;

	precisChannelList->number = ntohl(precisChannelList->number);
	for(i = 0; i < precisChannelList->number; i++) {
	    precisChannelList->channel[i].key = ntohl(precisChannelList->channel[i].key);
	    precisChannelList->channel[i].start_time = ntohl(precisChannelList->channel[i].start_time);
	    precisChannelList->channel[i].end_time = ntohl(precisChannelList->channel[i].end_time);

	    nmxp_data_to_str(str_start, precisChannelList->channel[i].start_time);
	    nmxp_data_to_str(str_end, precisChannelList->channel[i].end_time);

571
	    if(!nmxp_meta_chan_set_times(chan_list, precisChannelList->channel[i].key, precisChannelList->channel[i].start_time, precisChannelList->channel[i].end_time)) {
572
573
574
		nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CHANNEL, "Key %d not found for %s!\n",
			precisChannelList->channel[i].key,
			NMXP_LOG_STR(precisChannelList->channel[i].name));
575
576
577
	    }

	    /*
578
	    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_ANY, "%12d %12s %10d %10d %20s %20s\n",
579
		    precisChannelList->channel[i].key, NMXP_LOG_STR(precisChannelList->channel[i].name),
580
		    precisChannelList->channel[i].start_time, precisChannelList->channel[i].end_time,
581
		    NMXP_LOG_STR(str_start), NMXP_LOG_STR(str_end));
582
		    */
583
584
585
	}

	/* Receive Message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
586
	ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
587
	nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
588
589
590
    }


591
592
    if(flag_request_channelinfo) {
	for(iter = chan_list; iter != NULL; iter = iter->next) {
593

594
595
596
597
	    if(getChannelNumberFromKey(iter->key) == 0) {
		/* DAP Step 5: Send Data Request */
		channelInfoRequestBody.key = htonl(iter->key);
		channelInfoRequestBody.ignored = htonl(0);
598
		nmxp_sendMessage(naqssock, NMXP_MSG_CHANNELINFOREQUEST, &channelInfoRequestBody, sizeof(NMXP_CHANNELINFOREQUEST));
599

600
		/* DAP Step 6: Receive Data until receiving a Ready message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
601
		ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
602
		nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
603

604
605
606
		while(ret_sock == NMXP_SOCKET_OK   &&    type != NMXP_MSG_READY) {
		    channelInfo = buffer;
		    channelInfo->key = ntohl(channelInfo->key);
607

608
		    if(!nmxp_meta_chan_set_network(chan_list, channelInfo->key, channelInfo->network)) {
609
610
			nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_CHANNEL, "Key %d (%d) not found for %s!\n",
				iter->key, channelInfo->key, NMXP_LOG_STR(iter->name));
611
612
		    }
		    /* Receive Message */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
613
		    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length, 0, &recv_errno);
614
		    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_EXTRA, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
615
616
617
		}
	    }
	}
618
619
620
621
622
623
624
625
626
627
628
629
    }



    /* DAP Step 7: Repeat steps 5 and 6 for each data request */

    /* DAP Step 8: Send a Terminate message (optional) */
    nmxp_sendTerminateSubscription(naqssock, NMXP_SHUTDOWN_NORMAL, "Bye!");

    /* DAP Step 9: Close the socket */
    nmxp_closeSocket(naqssock);

630
    return chan_list;
631
632
}

633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648

int nmxp_raw_stream_seq_no_compare(const void *a, const void *b)
{       
    int ret = 0;
    NMXP_DATA_PROCESS **ppa = (NMXP_DATA_PROCESS **) a;
    NMXP_DATA_PROCESS **ppb = (NMXP_DATA_PROCESS **) b;
    NMXP_DATA_PROCESS *pa = *ppa;
    NMXP_DATA_PROCESS *pb = *ppb;

    if(pa && pb) {
	if(pa->seq_no > pb->seq_no) {
	    ret = 1;
	} else if (pa->seq_no < pb->seq_no) {
	    ret = -1;
	}
    } else {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
649
650
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_ANY,
		"nmxp_raw_stream_seq_no_compare() pa %s NULL and pb %s NULL\n", (pa)? "!=" : "=", (pb)? "!=" : "=");
651
652
653
654
655
    }

    return ret;
}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
656
void nmxp_raw_stream_init(NMXP_RAW_STREAM_DATA *raw_stream_buffer, int32_t max_tolerable_latency, int timeoutrecv) {
657
658
659
660
    int j;

    raw_stream_buffer->last_seq_no_sent = -1;
    raw_stream_buffer->last_sample_time = -1.0;
661
    /* TODO 
Matteo Quintiliani's avatar
Matteo Quintiliani committed
662
     * Suppose a packet can contain 1/4 secs of data (that is, minimum packet length is 0.25 secs of data) */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
663
664
    raw_stream_buffer->max_tolerable_latency = max_tolerable_latency;
    raw_stream_buffer->max_pdlist_items = max_tolerable_latency * 4;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
665
    raw_stream_buffer->timeoutrecv = timeoutrecv;
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
    raw_stream_buffer->n_pdlist = 0;
    raw_stream_buffer->pdlist = (NMXP_DATA_PROCESS **) malloc (raw_stream_buffer->max_pdlist_items * sizeof(NMXP_DATA_PROCESS *));
    for(j=0; j<raw_stream_buffer->max_pdlist_items; j++) {
	raw_stream_buffer->pdlist[j] = NULL;
    }

}


void nmxp_raw_stream_free(NMXP_RAW_STREAM_DATA *raw_stream_buffer) {
    int j;
    if(raw_stream_buffer->pdlist) {
	for(j=0; j<raw_stream_buffer->max_pdlist_items; j++) {
	    if(raw_stream_buffer->pdlist[j]) {
		free(raw_stream_buffer->pdlist[j]);
	    }
	}
	free(raw_stream_buffer->pdlist);
    }
}


Matteo Quintiliani's avatar
Matteo Quintiliani committed
688
int nmxp_raw_stream_manage(NMXP_RAW_STREAM_DATA *p, NMXP_DATA_PROCESS *a_pd, int (*p_func_pd[NMXP_MAX_FUNC_PD]) (NMXP_DATA_PROCESS *), int n_func_pd) {
689
690
691
692
693
694
695
696
697
    int ret = 0;
    int send_again = 1;
    int seq_no_diff;
    double time_diff;
    double latency = 0.0;
    int j=0, k=0;
    int i_func_pd;
    char str_time[200];
    NMXP_DATA_PROCESS *pd = NULL;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
698
    int y, w;
699
700
    int count_null_element = 0;
    char netstachan[100];
701

702
    /* Allocate pd copy value from a_pd */
703
    if(a_pd) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
	/*
	if(a_pd->packet_type == 33 || a_pd->packet_type == 97) {
	    nmxp_data_log(a_pd);
	}
	*/

	/* Allocate memory for pd and copy a_pd */
	pd = (NMXP_DATA_PROCESS *) malloc (sizeof(NMXP_DATA_PROCESS));
	memcpy(pd, a_pd, sizeof(NMXP_DATA_PROCESS));
	if(a_pd->length > 0) {
	    pd->buffer = malloc(pd->length);
	    memcpy(pd->buffer, a_pd->buffer, a_pd->length);
	} else {
	    pd->buffer = NULL;
	}
	if(a_pd->nSamp *  sizeof(int) > 0) {
	    pd->pDataPtr = (int *) malloc(a_pd->nSamp * sizeof(int));
	    memcpy(pd->pDataPtr, a_pd->pDataPtr, a_pd->nSamp * sizeof(int));
	} else {
	    pd->pDataPtr = NULL;
	}
725
726
    } else {
	nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
727
		"nmxp_raw_stream_manage() passing NMXP_DATA_PROCESS pointer equal to NULL\n");
728
    }
729
    /* From here, use only pd */
730
731

    /* First time */
732
    if(p->last_seq_no_sent == -1  &&  pd) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
733
734
735
736
737
738
739
	if(p->timeoutrecv == 0) {
	    p->last_seq_no_sent = pd->seq_no - 1;
	    p->last_sample_time = pd->time;
	} else {
	    p->last_seq_no_sent = 0;
	    p->last_sample_time = 0;
	}
740
	nmxp_data_to_str(str_time, pd->time);
741
	nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_RAWSTREAM,
742
743
744
745
746
		"%s.%s.%s [%d, %d] (%s + %.2f sec.) * First time nmxp_raw_stream_manage() * last_seq_no_sent=%d  last_sample_time=%.2f\n",
		NMXP_LOG_STR(pd->network), NMXP_LOG_STR(pd->station), NMXP_LOG_STR(pd->channel),
		pd->packet_type, pd->seq_no,
		NMXP_LOG_STR(str_time), (double) pd->nSamp / (double) pd->sampRate,
		 p->last_seq_no_sent, p->last_sample_time);
747
    }
Matteo Quintiliani's avatar
Matteo Quintiliani committed
748

749
750
751
    if(p->n_pdlist > 0) {
	latency = nmxp_data_latency(p->pdlist[0]);
    }
752

753
754
755
    /* Add pd and sort array, in case handle the first item */
    if( (p->n_pdlist >= p->max_pdlist_items || latency >= p->max_tolerable_latency)
	    && p->timeoutrecv <= 0 ) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
756

757
	/* Supposing p->pdlist is ordered, handle the first item and over write it.  */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
758
759
760
761
762
763
	if(p->n_pdlist > 0) {
	    seq_no_diff = p->pdlist[0]->seq_no - p->last_seq_no_sent;
	    time_diff = p->pdlist[0]->time - p->last_sample_time;
	    latency = nmxp_data_latency(p->pdlist[0]);
	    nmxp_data_to_str(str_time, p->pdlist[0]->time);
	    if( seq_no_diff > 0) {
764
		nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
765
766
767
768
769
770
			"%s.%s.%s [%d, %d] (%s + %.2f sec.) * Force handling packet * n_pdlist=%d  seq_no_diff=%d  time_diff=%.2fs  lat. %.1fs!\n",
			NMXP_LOG_STR(p->pdlist[0]->network), NMXP_LOG_STR(p->pdlist[0]->station), NMXP_LOG_STR(p->pdlist[0]->channel),
			p->pdlist[0]->packet_type, p->pdlist[0]->seq_no,
			NMXP_LOG_STR(str_time), (double) p->pdlist[0]->nSamp / (double) p->pdlist[0]->sampRate,
			p->n_pdlist,
			seq_no_diff, time_diff, latency);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
771
772
773
774
775
776
777
		for(i_func_pd=0; i_func_pd<n_func_pd; i_func_pd++) {
		    (*p_func_pd[i_func_pd])(p->pdlist[0]);
		}
		p->last_seq_no_sent = (p->pdlist[0]->seq_no);
		p->last_sample_time = (p->pdlist[0]->time + ((double) p->pdlist[0]->nSamp / (double) p->pdlist[0]->sampRate ));
	    } else {
		/* It should not occur */
778
779
780
781
782
783
784
		nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_RAWSTREAM,
			"%s.%s.%s [%d, %d] (%s + %.2f sec.) * SHOULD NOT OCCUR packet discarded * n_pdlist=%d  seq_no_diff=%d  time_diff=%.2fs  lat. %.1fs!\n",
			NMXP_LOG_STR(p->pdlist[0]->network), NMXP_LOG_STR(p->pdlist[0]->station), NMXP_LOG_STR(p->pdlist[0]->channel),
			p->pdlist[0]->packet_type, p->pdlist[0]->seq_no,
			NMXP_LOG_STR(str_time), (double) p->pdlist[0]->nSamp / (double) p->pdlist[0]->sampRate,
			p->n_pdlist,
			seq_no_diff, time_diff, latency);
785
786
	    }

Matteo Quintiliani's avatar
Matteo Quintiliani committed
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
	    /* Free handled packet */
	    if(p->pdlist[0]->buffer) {
		free(p->pdlist[0]->buffer);
		p->pdlist[0]->buffer = NULL;
	    }
	    if(p->pdlist[0]->pDataPtr) {
		free(p->pdlist[0]->pDataPtr);
		p->pdlist[0]->pDataPtr = NULL;
	    }
	    if(p->pdlist[0]) {
		free(p->pdlist[0]);
		p->pdlist[0] = NULL;
	    }
	    if(pd) {
		p->pdlist[0] = pd;
	    }
	} else {
	    if(pd) {
		p->n_pdlist = 1;
		p->pdlist[0] = pd;
	    }
808
809
	}
    } else {
810
811
812
	if(pd) {
	    p->pdlist[p->n_pdlist++] = pd;
	}
813
    }
814
815
816
817
818

    /* Check if some element in pdlist is NULL and remove it */
    y=0;
    while(y < p->n_pdlist) {
	if(p->pdlist[y] == NULL) {
819
	    count_null_element++;
820
821
822
823
824
825
826
827
828
829
	    /* Shift array */
	    for(w=y+1; w < p->n_pdlist; w++) {
		p->pdlist[w-1] = p->pdlist[w];
	     }
	    p->n_pdlist--;
	} else {
	    y++;
	}
    }

830
831
832
833
834
835
836
837
838
839
840
841
842
843
    if(count_null_element > 0) {
	if(p->n_pdlist > 0) {
	    snprintf(netstachan, 100, "%s.%s.%s",
		    NMXP_LOG_STR(p->pdlist[0]->network),
		    NMXP_LOG_STR(p->pdlist[0]->station),
		    NMXP_LOG_STR(p->pdlist[0]->channel));
	} else {
	    strncpy(netstachan, "Unknown", 100);
	}
	nmxp_log(NMXP_LOG_ERR, NMXP_LOG_D_ANY,
		"nmxp_raw_stream_manage() %d NULL elements in pdlist for %s.\n",
		count_null_element, netstachan);
    }

844
    /* Sort array */
845
846
    qsort(p->pdlist, p->n_pdlist, sizeof(NMXP_DATA_PROCESS *), nmxp_raw_stream_seq_no_compare);

847
    /* TODO Check for packet duplication in pd->pdlist*/
848
849

    /* Print array, only for debugging */
850
    /*
851
852
853
    if(p->n_pdlist > 1) {
	int y = 0;
	for(y=0; y < p->n_pdlist; y++) {
854
855
856
857
858
859
860
	    nmxp_data_to_str(str_time, p->pdlist[y]->time);
	    nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
		    "%s.%s.%s [%d, %d] (%s + %.2f sec.) * %02d n_pdlist=%d\n",
		    NMXP_LOG_STR(p->pdlist[y]->network), NMXP_LOG_STR(p->pdlist[y]->station), NMXP_LOG_STR(p->pdlist[y]->channel),
		    p->pdlist[y]->packet_type, p->pdlist[y]->seq_no,
		    NMXP_LOG_STR(str_time), (double) p->pdlist[y]->nSamp / (double) p->pdlist[y]->sampRate,
		    y, p->n_pdlist);
861
862
	}
    }
863
    */
864

865
    /* Condition for time-out (pd is NULL) */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
866
    if(!pd && p->n_pdlist > 0) {
867
868
869
870
871
872
873
874
875
876
877
	/* Log before changing values */
	nmxp_data_to_str(str_time, p->pdlist[0]->time);
	nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
		"%s.%s.%s [%d, %d] (%s + %.2f sec.) * pd is NULL and n_pdlist = %d > 0 *  last_seq_no_sent=%d, last_sample_time=%.2f\n",
		NMXP_LOG_STR(p->pdlist[0]->network), NMXP_LOG_STR(p->pdlist[0]->station), NMXP_LOG_STR(p->pdlist[0]->channel),
		p->pdlist[0]->packet_type, p->pdlist[0]->seq_no,
		NMXP_LOG_STR(str_time), (double) p->pdlist[0]->nSamp / (double) p->pdlist[0]->sampRate,
		p->n_pdlist,
		p->last_seq_no_sent, p->last_sample_time);

	/* Changing values */
Matteo Quintiliani's avatar
Matteo Quintiliani committed
878
879
880
	p->last_seq_no_sent = p->pdlist[0]->seq_no - 1;
	p->last_sample_time = p->pdlist[0]->time;
    }
881
882
883
884
885
886
887
888
889
890
891

    /* Manage array and execute func_pd() */
    j=0;
    send_again = 1;
    while(send_again  &&  j < p->n_pdlist) {
	send_again = 0;
	seq_no_diff = p->pdlist[j]->seq_no - p->last_seq_no_sent;
	time_diff = p->pdlist[j]->time - p->last_sample_time;
	latency = nmxp_data_latency(p->pdlist[j]);
	nmxp_data_to_str(str_time, p->pdlist[j]->time);
	if(seq_no_diff <= 0) {
892
	    /* Duplicated packets: Discarded */
893
	    nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
894
895
896
897
898
		    "%s.%s.%s [%d, %d] (%s + %.2f sec.) * Packet discarded * seq_no_diff=%d  time_diff=%.2fs  lat %.1fs\n",
		    NMXP_LOG_STR(p->pdlist[j]->network), NMXP_LOG_STR(p->pdlist[j]->station), NMXP_LOG_STR(p->pdlist[j]->channel),
		    p->pdlist[j]->packet_type, p->pdlist[j]->seq_no, 
		    NMXP_LOG_STR(str_time), (double) p->pdlist[j]->nSamp / (double) p->pdlist[j]->sampRate,
		    seq_no_diff, time_diff, latency);
899
900
901
	    send_again = 1;
	    j++;
	} else if(seq_no_diff == 1) {
902
	    /* Handle current packet j */
903
904
905
906
	    for(i_func_pd=0; i_func_pd<n_func_pd; i_func_pd++) {
		(*p_func_pd[i_func_pd])(p->pdlist[j]);
	    }
	    if(time_diff > TIME_TOLLERANCE || time_diff < -TIME_TOLLERANCE) {
907
		nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
908
909
910
911
912
913
			"%s.%s.%s [%d, %d] (%s + %.2f sec.) * Time is not correct * last_seq_no_sent=%d  seq_no_diff=%d  time_diff=%.2fs  lat. %.1fs\n",
		    NMXP_LOG_STR(p->pdlist[j]->network), NMXP_LOG_STR(p->pdlist[j]->station), NMXP_LOG_STR(p->pdlist[j]->channel), 
		    p->pdlist[j]->packet_type, p->pdlist[j]->seq_no,
		    str_time, (double) p->pdlist[j]->nSamp /  (double) p->pdlist[j]->sampRate,
		    p->last_seq_no_sent,
		    seq_no_diff, time_diff, latency);
914
915
916
917
918
919
	    }
	    p->last_seq_no_sent = p->pdlist[j]->seq_no;
	    p->last_sample_time = (p->pdlist[j]->time + ((double) p->pdlist[j]->nSamp / (double) p->pdlist[j]->sampRate ));
	    send_again = 1;
	    j++;
	} else {
920
	    nmxp_log(NMXP_LOG_WARN, NMXP_LOG_D_RAWSTREAM,
921
922
923
924
925
926
		    "%s.%s.%s [%d, %d] (%s + %.2f sec.) * seq_no_diff=%d > 1 * last_seq_no_sent=%d  j=%d  n_pdlist=%2d  time_diff=%.2fs  lat. %.1fs\n",
		    NMXP_LOG_STR(p->pdlist[j]->network), NMXP_LOG_STR(p->pdlist[j]->station), NMXP_LOG_STR(p->pdlist[j]->channel), 
		    p->pdlist[j]->packet_type, p->pdlist[j]->seq_no,
		    str_time, (double) p->pdlist[j]->nSamp /  (double) p->pdlist[j]->sampRate,
		    seq_no_diff, p->last_seq_no_sent, j, p->n_pdlist,
		    time_diff, latency);
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
	}
    }

    /* Shift and free j handled elements */
    if(j > 0) {
	for(k=0; k < p->n_pdlist; k++) {
	    if(k < j) {
		if(p->pdlist[k]->buffer) {
		    free(p->pdlist[k]->buffer);
		    p->pdlist[k]->buffer = NULL;
		}
		if(p->pdlist[k]->pDataPtr) {
		    free(p->pdlist[k]->pDataPtr);
		    p->pdlist[k]->pDataPtr = NULL;
		}
		if(p->pdlist[k]) {
		    free(p->pdlist[k]);
		    p->pdlist[k] = NULL;
		}
	    }
	    if(k + j < p->n_pdlist) {
		p->pdlist[k] = p->pdlist[k+j];
	    } else {
		p->pdlist[k] = NULL;
	    }
	}
	p->n_pdlist = p->n_pdlist - j;
    }

956
957
958
    /* TOREMOVE
    nmxp_log(NMXP_LOG_NORM, NMXP_LOG_D_RAWSTREAM, "j=%d  p->n_pdlist=%d FINAL\n", j, p->n_pdlist);
       */
959
960
961
962

    return ret;
}

963
964
965
966
967
968
969

/* TODO */
int nmxp_raw_stream_manage_flush(NMXP_RAW_STREAM_DATA *p, int (*p_func_pd[NMXP_MAX_FUNC_PD]) (NMXP_DATA_PROCESS *), int n_func_pd) {
    int ret = 0;

    return ret;
}