nmxp.c 8.58 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
34
35
36
37
    void *buffer;
    uint32_t length;

    *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;
}


Matteo Quintiliani's avatar
Matteo Quintiliani committed
62
int nmxp_sendAddTimeSeriesChannel(int isock, NMXP_CHAN_LIST *channelList, uint32_t shortTermCompletion, uint32_t out_format, NMXP_BUFFER_FLAG buffer_flag) {
63
64
65
66
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
    int ret;
    uint32_t buffer_length = 16 + (4 * channelList->number); 
    char *buffer = malloc(buffer_length);
    uint32_t app, i, disp;

    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
int nmxp_receiveData(int isock, NMXP_CHAN_LIST *channelList,
102
	int (*func_processData)(NMXP_DATA_PROCESS *pd)
103
	) {
104
105
    int ret;

Matteo Quintiliani's avatar
Matteo Quintiliani committed
106
    NMXP_MSG_SERVER type;
107
108
    void *buffer;
    uint32_t length;
109
    NMXP_DATA_PROCESS *pd = NULL;
110

111
    ret = nmxp_receiveMessage(isock, &type, &buffer, &length);
112

113
    if(type == NMXP_MSG_COMPRESSED) {
114
	pd = nmxp_processCompressedDataFunc(buffer, length, channelList);
115
    } else if(type == NMXP_MSG_DECOMPRESSED) {
116
	pd = nmxp_processDecompressedDataFunc(buffer, length, channelList);
117
    } else {
118
	nmxp_log(1, 0, "Type %d is not NMXP_MSG_COMPRESSED or NMXP_MSG_DECOMPRESSED!\n", type);
119
120
    }

121
122
123
124
    if(pd) {
	func_processData(pd);
    }

Matteo Quintiliani's avatar
Matteo Quintiliani committed
125
126
127
128
    if(buffer) {
	free(buffer);
    }

129
130
131
    return ret;
}

132
133
134
135
136

int nmxp_sendConnectRequest(int isock, char *naqs_username, char *naqs_password, uint32_t connection_time) {
    int ret;
    char crc32buf[100];
    NMXP_CONNECT_REQUEST connectRequest;
137
    int naqs_username_length, naqs_password_length;
138

139
140
141
142
143
    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
144
    } else {
145
	strcpy(connectRequest.username, naqs_username);
Matteo Quintiliani's avatar
Matteo Quintiliani committed
146
    }
147
148
149
    connectRequest.version = htonl(0);
    connectRequest.connection_time = htonl(connection_time);

150
    if(naqs_username_length == 0  &&  naqs_password_length == 0 ) {
151
	sprintf(crc32buf, "%d%d", connectRequest.version, connection_time);
152
    } else if(naqs_username_length != 0  &&  naqs_password_length != 0 ) {
153
154
	sprintf(crc32buf, "%s%d%d%s", naqs_username, connectRequest.version,
		connection_time, naqs_password);
155
    } else if(naqs_username_length != 0 ) {
156
	sprintf(crc32buf, "%s%d%d", naqs_username, connectRequest.version, connection_time);
157
    } else if(naqs_password_length != 0 ) {
158
159
160
161
162
163
164
	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) {
165
	nmxp_log(0, 1, "Send a ConnectRequest crc32buf = (%s), crc32 = %d\n", crc32buf, crc32);
166
    } else {
167
	nmxp_log(1, 0, "Send a ConnectRequest.\n");
168
169
170
171
172
173
174
175
176
177
    }

    return ret;
}


int nmxp_readConnectionTime(int isock, uint32_t *connection_time) {
    int ret;
    ret = nmxp_recv_ctrl(isock, connection_time, sizeof(connection_time));
    *connection_time = ntohl(*connection_time);
178
    nmxp_log(0, 1, "Read connection time from socket %d.\n", *connection_time);
179
    if(ret != NMXP_SOCKET_OK) {
180
	nmxp_log(1, 0, "Read connection time from socket.\n");
181
182
183
184
185
    }
    return ret;
}


186
int nmxp_waitReady(int isock) {
187
188
189
190
191
192
193
194
195
196
197
    int times = 0;
    int rc = NMXP_SOCKET_OK;
    unsigned long signature;
    unsigned long type = 0;
    unsigned long length;

    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) {
198
	    nmxp_log(0, 1, "signature is equal to zero. receive again.\n");
199
200
201
202
203
204
205
206
207
208
209
210
	    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);
	    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) {
211
	    nmxp_log(0, 1, "type is not READY. type = %d\n", type);
212
213
214
215
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
256
257
258
259
260
261
262
263
264
265
266
267
268
269
	    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;
}


int nmxp_sendDataRequest(int isock, uint32_t key, uint32_t start_time, uint32_t end_time) {
    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;
}


270
271
272
273
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
310
311
312
313
314
315
316
317
318
319
320
321
322
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 {
		 printf("Error on receiveChannelList()\n");
	     }
	} else {
	    printf("Error on sendConnect()\n");
	}

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

    if(channelList) {
	free(channelList);
    }

    return channelList_subset;
}