1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
#include <zmq.h>
#include <assert.h>
#include <time.h>
#include <dirent.h>
#include "utils.h"
// Global:
char *__logdir;
char *__key;
char *__interface;
char *__ip;
int __loginterval;
int __port;
void publish(void *publisher, char *filepath, char* client, long int interval);
int main (int argc, char *argv [])
{
if(argc != 6){
printf("Usage: %s <abslogdir> <loginterval> <ip> <port> <key>",argv[0]);
exit(1);
}
//----- Init global variables
__logdir=argv[1];
__loginterval=atoi(argv[2]);
__ip=argv[3];
__port=atoi(argv[4]);
__key=argv[5];
//----- Prepare our context and publisher
void *context = zmq_ctx_new ();
void *publisher = zmq_socket (context, ZMQ_PUB);
char bindto[30];
sprintf(bindto,"tcp://%s:%d",__ip,__port);
int rc = zmq_connect (publisher, bindto);
if(rc!=0){
printf("Failed to connect to %s\n",bindto);
exit(1);
}
//----- Start publisher
struct dirent *de; // Pointer for directory entry
while(1){
int interval=INTERVAL(__loginterval);
int interval_next=INTERVAL_NEXT(__loginterval);
DIR *dr = opendir(__logdir);
if(dr !=NULL){
while ((de = readdir(dr)) != NULL){
if(strcmp(de->d_name,".") && strcmp(de->d_name,"..")){
char *client=de->d_name;
char logfile[255];
char logfile_next[255];
sprintf(logfile,"%s/%s/%ld",__logdir,client,interval);
sprintf(logfile_next,"%s/%s/%ld",__logdir,client,interval_next);
// As long as next logfile is not available, we should wait
// for sending the current one
printf("Waiting for %s logger measurements...\n",client);
while(!FILE_EXISTS(logfile_next)){
sleep(1);
}
// Send current one
if(FILE_EXISTS(logfile)){
publish(publisher,logfile,client,interval);
remove(logfile);
}
}
}
closedir(dr);
}
else {
sleep(1);
}
}
zmq_close (publisher);
zmq_ctx_destroy (context);
return 0;
}
void publish(void *publisher, char *filepath, char* client, long int interval){
printf("Publishing %s measurements\n",client);
char buffer[ZMQ_MSG_SIZE];
sprintf(buffer,"%s\n%s\n%s\n%ld\n",ZMQ_TOKEN,__key,client,interval);
int msglen=strlen(buffer);
FILE *fptr;
char * line = NULL;
size_t len = 0;
ssize_t read;
fptr=fopen(filepath,"r");
while ((read = getline(&line, &len, fptr)) != -1) {
if((read+msglen) <ZMQ_MSG_SIZE){
strcat(buffer,line);
msglen+=read;
} else {
zmq_send (publisher, buffer, msglen, 0);
sprintf(buffer,"%s\n%s\n%s\n%ld\n",ZMQ_TOKEN,__key,client,interval);
strcat(buffer,line);
msglen=strlen(buffer);
}
}
fclose(fptr);
zmq_send (publisher, buffer, msglen, 0);
}
|
