nmxp.c 15 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
10
11
12
13
 *
 * Author:
 * 	Matteo Quintiliani
 * 	Istituto Nazionale di Geofisica e Vulcanologia - Italy
 *	quintiliani@ingv.it
 *
 */

#include "nmxp.h"

Matteo Quintiliani's avatar
Matteo Quintiliani committed
14
15
#include "config.h"

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

20
int nmxp_sendConnect(int isock) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
21
    return nmxp_sendMessage(isock, NMXP_MSG_CONNECT, NULL, 0);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
22
23
}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
24
25
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
26
27
}

Matteo Quintiliani's avatar
Matteo Quintiliani committed
28
int nmxp_receiveChannelList(int isock, NMXP_CHAN_LIST **pchannelList) {
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
29
30
31
    int ret;
    int i;

Matteo Quintiliani's avatar
Matteo Quintiliani committed
32
    NMXP_MSG_SERVER type;
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
33
    void *buffer;
34
    int32_t length;
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
35
36
37

    *pchannelList = NULL;

38
    ret = nmxp_receiveMessage(isock, &type, &buffer, &length);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
39

Matteo Quintiliani's avatar
Matteo Quintiliani committed
40
41
    if(type != NMXP_MSG_CHANNELLIST) {
	nmxp_log(1, 0, "Type %d is not NMXP_MSG_CHANNELLIST!\n", type);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
42
43
44
45
46
    } else {

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

Matteo Quintiliani's avatar
Matteo Quintiliani committed
47
	nmxp_log(0, 1, "number of channels %d\n", (*pchannelList)->number);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
48
49
50
51
52
	
	// TODO check

	for(i=0; i < (*pchannelList)->number; i++) {
	    (*pchannelList)->channel[i].key = ntohl((*pchannelList)->channel[i].key);
53
	    nmxp_log(0, 1, "%12d %s\n", (*pchannelList)->channel[i].key, (*pchannelList)->channel[i].name);
Matteo Quintiliani's avatar
Start    
Matteo Quintiliani committed
54
55
56
57
58
59
60
61
	}

    }

    return ret;
}


62
int nmxp_sendAddTimeSeriesChannel(int isock, NMXP_CHAN_LIST *channelList, int32_t shortTermCompletion, int32_t out_format, NMXP_BUFFER_FLAG buffer_flag) {
63
    int ret;
64
    int32_t buffer_length = 16 + (4 * channelList->number); 
65
    char *buffer = malloc(buffer_length);
66
    int32_t app, i, disp;
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

    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
92
    ret = nmxp_sendMessage(isock, NMXP_MSG_ADDTIMESERIESCHANNELS, buffer, buffer_length);
93
94
95
96
97
98
99
100

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


101
NMXP_DATA_PROCESS *nmxp_receiveData(int isock, NMXP_CHAN_LIST *channelList, const char *network_code) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
102
    NMXP_MSG_SERVER type;
103
    void *buffer = NULL;
104
    int32_t length;
105
    NMXP_DATA_PROCESS *pd = NULL;
106

107
108
109
    if(nmxp_receiveMessage(isock, &type, &buffer, &length) == NMXP_SOCKET_OK) {
	if(type == NMXP_MSG_COMPRESSED) {
	    nmxp_log(0, 1, "Type %d is NMXP_MSG_COMPRESSED!\n", type);
110
	    pd = nmxp_processCompressedData(buffer, length, channelList, network_code);
111
112
	} else if(type == NMXP_MSG_DECOMPRESSED) {
	    nmxp_log(0, 1, "Type %d is NMXP_MSG_DECOMPRESSED!\n", type);
113
	    pd = nmxp_processDecompressedData(buffer, length, channelList, network_code);
114
115
116
	} else {
	    nmxp_log(1, 0, "Type %d is not NMXP_MSG_COMPRESSED or NMXP_MSG_DECOMPRESSED!\n", type);
	}
Matteo Quintiliani's avatar
Matteo Quintiliani committed
117
118
    }

119
    return pd;
120
121
}

122

123
int nmxp_sendConnectRequest(int isock, char *naqs_username, char *naqs_password, int32_t connection_time) {
124
125
126
    int ret;
    char crc32buf[100];
    NMXP_CONNECT_REQUEST connectRequest;
127
    int naqs_username_length, naqs_password_length;
128

129
130
131
132
133
    naqs_username_length = (naqs_username)? strlen(naqs_username) : 0;
    naqs_password_length = (naqs_password)? strlen(naqs_password) : 0;

    if(naqs_username_length == 0) {
	connectRequest.username[0] = 0;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
134
    } else {
135
	strcpy(connectRequest.username, naqs_username);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
136
    }
137
138
139
    connectRequest.version = htonl(0);
    connectRequest.connection_time = htonl(connection_time);

140
    if(naqs_username_length == 0  &&  naqs_password_length == 0 ) {
141
	sprintf(crc32buf, "%d%d", connectRequest.version, connection_time);
142
    } else if(naqs_username_length != 0  &&  naqs_password_length != 0 ) {
143
144
	sprintf(crc32buf, "%s%d%d%s", naqs_username, connectRequest.version,
		connection_time, naqs_password);
145
    } else if(naqs_username_length != 0 ) {
146
	sprintf(crc32buf, "%s%d%d", naqs_username, connectRequest.version, connection_time);
147
    } else if(naqs_password_length != 0 ) {
148
149
150
151
152
153
154
	sprintf(crc32buf, "%d%d%s", connectRequest.version, connection_time, naqs_password);
    }
    connectRequest.crc32 = htonl(crc32((unsigned char *) crc32buf, strlen(crc32buf)));

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

    if(ret == NMXP_SOCKET_OK) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
155
	nmxp_log(0, 1, "Send a ConnectRequest crc32buf = (%s), crc32 = %d\n", crc32buf, connectRequest.crc32);
156
    } else {
157
	nmxp_log(1, 0, "Send a ConnectRequest.\n");
158
159
160
161
162
163
    }

    return ret;
}


164
int nmxp_readConnectionTime(int isock, int32_t *connection_time) {
165
    int ret;
166
    ret = nmxp_recv_ctrl(isock, connection_time, sizeof(int32_t));
167
    *connection_time = ntohl(*connection_time);
168
    nmxp_log(0, 1, "Read connection time from socket %d.\n", *connection_time);
169
    if(ret != NMXP_SOCKET_OK) {
170
	nmxp_log(1, 0, "Read connection time from socket.\n");
171
172
173
174
175
    }
    return ret;
}


176
int nmxp_waitReady(int isock) {
177
178
    int times = 0;
    int rc = NMXP_SOCKET_OK;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
179
180
181
    int32_t signature;
    int32_t type = 0;
    int32_t length;
182
183
184
185
186
187

    while(rc == NMXP_SOCKET_OK  &&  type != NMXP_MSG_READY) {
	rc = nmxp_recv_ctrl(isock, &signature, sizeof(signature));
	if(rc != NMXP_SOCKET_OK) return rc;
	signature = ntohl(signature);
	if(signature == 0) {
188
	    nmxp_log(0, 1, "signature is equal to zero. receive again.\n");
189
190
191
192
193
	    rc = nmxp_recv_ctrl(isock, &signature, sizeof(signature));
	    signature = ntohl(signature);
	}
	if(signature != NMX_SIGNATURE) {
	    nmxp_log(1, 0, "signature is not valid. signature = %d\n", signature);
194
195
196
197
198
199
200
201
202
203
204
205
206
207
	    if(signature == 200) {
		    int32_t err_length;
		    int32_t err_reason;
		    char err_buff[200];
		    rc = nmxp_recv_ctrl(isock, &err_length, sizeof(err_length));
		    err_length = ntohl(err_length);
		    rc = nmxp_recv_ctrl(isock, &err_reason, sizeof(err_reason));
		    err_reason = ntohl(err_reason);
		    if(err_length > 4) {
			    rc = nmxp_recv_ctrl(isock, err_buff, err_length-4);
			    err_buff[err_length] = 0;
		    }
		    nmxp_log(1, 0, "TerminateMessage from Server: %s (%d).\n", err_buff, err_reason);
	    }
208
209
210
211
212
213
214
	    return NMXP_SOCKET_ERROR;
	}

	rc = nmxp_recv_ctrl(isock, &type, sizeof(type));
	if(rc != NMXP_SOCKET_OK) return rc;
	type = ntohl(type);
	if(type != NMXP_MSG_READY) {
215
	    nmxp_log(0, 1, "type is not READY. type = %d\n", type);
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
	    rc = nmxp_recv_ctrl(isock, &length, sizeof(length));
	    if(rc != NMXP_SOCKET_OK) return rc;
	    length = ntohl(length);
	    if(length > 0) {
		if(length == 4) {
		    int32_t app;
		    rc = nmxp_recv_ctrl(isock, &app, length);
		    if(rc != NMXP_SOCKET_OK) return rc;
		    app = ntohl(app);
		    nmxp_log(0, 1, "value = %d\n", app);
		} else {
		    char *buf_app = (char *) malloc(sizeof(char) * length);
		    rc = nmxp_recv_ctrl(isock, buf_app, length);
		    if(buf_app) {
			free(buf_app);
		    }
		}
	    }
	} else {
	    rc = nmxp_recv_ctrl(isock, &length, sizeof(length));
	    if(rc != NMXP_SOCKET_OK) return rc;
	    length = ntohl(length);
	    if(length != 0) {
		nmxp_log(1, 0, "length is not equal to zero. length = %d\n", length);
		return NMXP_SOCKET_ERROR;
	    }
	}

	times++;
	if(times > 10) {
	    nmxp_log(1, 0, "waiting_ready_message. times > 10\n");
	    rc = NMXP_SOCKET_ERROR;
	}

    }

    return rc;
}


256
int nmxp_sendDataRequest(int isock, int32_t key, int32_t start_time, int32_t end_time) {
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
    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) {
	nmxp_log(1,0, "Send a Request message\n");
    }

    return ret;
}


274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
NMXP_CHAN_LIST *nmxp_getAvailableChannelList(char * hostname, int portnum, NMXP_DATATYPE datatype) {
    int naqssock;
    NMXP_CHAN_LIST *channelList = NULL, *channelList_subset = NULL;
    int i;

    // 1. Open a socket
    naqssock = nmxp_openSocket(hostname, portnum);

    if(naqssock != NMXP_SOCKET_ERROR) {

	// 2. Send a Connect
	if(nmxp_sendConnect(naqssock) == NMXP_SOCKET_OK) {

	    // 3. Receive ChannelList
	     if(nmxp_receiveChannelList(naqssock, &channelList) == NMXP_SOCKET_OK) {

		 channelList_subset = nmxp_chan_getType(channelList, datatype);
		 nmxp_log(0, 1, "%d / %d\n", channelList_subset->number, channelList->number);

		 // nmxp_chan_sortByKey(channelList_subset);
		 nmxp_chan_sortByName(channelList_subset);

		 for(i=0; i < channelList_subset->number; i++) {
		     nmxp_log(0, 1, "%12d %s\n", channelList_subset->channel[i].key, channelList_subset->channel[i].name);
		 }

		 // 4. Send a Request Pending (optional)

		 // 5. Send AddChannels

		 // 6. Repeat until finished: receive and handle packets

		 // 7. Send Terminate Subscription
		 nmxp_sendTerminateSubscription(naqssock, NMXP_SHUTDOWN_NORMAL, "Good Bye!");

	     } else {
310
		 nmxp_log(1, 0, "Error on receiveChannelList()\n");
311
312
	     }
	} else {
313
	    nmxp_log(1, 0, "Error on sendConnect()\n");
314
315
316
317
318
319
320
321
322
323
324
325
326
	}

	// 8. Close the socket
	nmxp_closeSocket(naqssock);
    }

    if(channelList) {
	free(channelList);
    }

    return channelList_subset;
}

327

328
NMXP_META_CHAN_LIST *nmxp_getMetaChannelList(char * hostname, int portnum, NMXP_DATATYPE datatype, int flag_request_channelinfo) {
329
330
    int naqssock;
    NMXP_CHAN_PRECISLIST *precisChannelList = NULL;
331
332
333
334
    NMXP_CHAN_LIST *channelList = NULL;
    NMXP_META_CHAN_LIST *chan_list = NULL;
    NMXP_META_CHAN_LIST *iter = NULL;
    int i = 0;
335
    int32_t connection_time;
336
337
338
    char *datas_username = NULL, *datas_password = NULL;
    int ret_sock;

339
    
340
341
    NMXP_MSG_SERVER type;
    void *buffer = NULL;
342
    int32_t length;
Matteo Quintiliani's avatar
Matteo Quintiliani committed
343
    NMXP_MSGBODY_PRECISLISTREQUEST precisListRequestBody;
344
345
    NMXP_MSGBODY_CHANNELINFOREQUEST channelInfoRequestBody;
    NMXP_MSGBODY_CHANNELINFORESPONSE *channelInfo;
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375

    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) {
	nmxp_log(1, 0, "Error opening socket!\n");
	return NULL;
    }

    /* DAP Step 2: Read connection time */
    if(nmxp_readConnectionTime(naqssock, &connection_time) != NMXP_SOCKET_OK) {
	nmxp_log(1, 0, "Error reading connection time from server!\n");
	return NULL;
    }

    /* DAP Step 3: Send a ConnectRequest */
    if(nmxp_sendConnectRequest(naqssock, datas_username, datas_password, connection_time) != NMXP_SOCKET_OK) {
	nmxp_log(1, 0, "Error sending connect request!\n");
	return NULL;
    }

    /* DAP Step 4: Wait for a Ready message */
    if(nmxp_waitReady(naqssock) != NMXP_SOCKET_OK) {
	nmxp_log(1, 0, "Error waiting Ready message!\n");
	return NULL;
    }

376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391



    /* DAP Step 5: Send Data Request */
    nmxp_sendHeader(naqssock, NMXP_MSG_CHANNELLISTREQUEST, 0);
    /* DAP Step 6: Receive Data until receiving a Ready message */
    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
    nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);

    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);
392
	    if(getDataTypeFromKey(channelList->channel[i].key) == datatype) {
Matteo Quintiliani's avatar
Matteo Quintiliani committed
393
		nmxp_meta_chan_add(&chan_list, channelList->channel[i].key, channelList->channel[i].name, 0, 0, NULL, NMXP_META_SORT_NAME);
394
	    }
395
396
397
398
399
400
401
402
	}

	/* Receive Message */
	ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
	nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
    }


403
404
405
406
    /* DAP Step 5: Send Data Request */
    precisListRequestBody.instr_id = htonl(-1);
    precisListRequestBody.datatype = htonl(NMXP_DATA_TIMESERIES);
    precisListRequestBody.type_of_channel = htonl(-1);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
407
    nmxp_sendMessage(naqssock, NMXP_MSG_PRECISLISTREQUEST, &precisListRequestBody, sizeof(NMXP_MSGBODY_PRECISLISTREQUEST));
408

409
410
411

    /* DAP Step 6: Receive Data until receiving a Ready message */
    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
412
    nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
413
414
415
416
417
418
419
420
421
422
423
424
425

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

426
427
428
429
430
	    if(!nmxp_meta_chan_set_times(chan_list, precisChannelList->channel[i].key, precisChannelList->channel[i].start_time, precisChannelList->channel[i].end_time)) {
		nmxp_log(1, 0, "Key %d not found for %s!\n", precisChannelList->channel[i].key, precisChannelList->channel[i].name);
	    }

	    /*
431
432
433
434
	    nmxp_log(0, 0, "%12d %12s %10d %10d %20s %20s\n",
		    precisChannelList->channel[i].key, precisChannelList->channel[i].name,
		    precisChannelList->channel[i].start_time, precisChannelList->channel[i].end_time,
		    str_start, str_end);
435
		    */
436
437
438
439
	}

	/* Receive Message */
	ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
440
441
442
443
	nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
    }


444
445
    if(flag_request_channelinfo) {
	for(iter = chan_list; iter != NULL; iter = iter->next) {
446

447
448
449
450
451
	    if(getChannelNumberFromKey(iter->key) == 0) {
		/* DAP Step 5: Send Data Request */
		channelInfoRequestBody.key = htonl(iter->key);
		channelInfoRequestBody.ignored = htonl(0);
		nmxp_sendMessage(naqssock, NMXP_MSG_CHANNELINFOREQUEST, &channelInfoRequestBody, sizeof(NMXP_MSGBODY_CHANNELINFOREQUEST));
452

453
454
455
		/* DAP Step 6: Receive Data until receiving a Ready message */
		ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
		nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
456

457
458
459
		while(ret_sock == NMXP_SOCKET_OK   &&    type != NMXP_MSG_READY) {
		    channelInfo = buffer;
		    channelInfo->key = ntohl(channelInfo->key);
460

461
462
463
464
465
466
		    if(!nmxp_meta_chan_set_network(chan_list, channelInfo->key, channelInfo->network)) {
			nmxp_log(1, 0, "Key %d (%d) not found for %s!\n", iter->key, channelInfo->key, iter->name);
		    }
		    /* Receive Message */
		    ret_sock = nmxp_receiveMessage(naqssock, &type, &buffer, &length);
		    nmxp_log(0, 1, "ret_sock = %d, type = %d, length = %d\n", ret_sock, type, length);
467
468
469
		}
	    }
	}
470
471
472
473
474
475
476
477
478
479
480
481
    }



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

482
483
484
    nmxp_meta_chan_print(chan_list);

    return chan_list;
485
486
}