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
|
// Weather update server
// Binds PUB socket to tcp://*:5556
// Publishes random weather updates
#include <zmq.h>
#include <assert.h>
#include <time.h>
#include <dirent.h>
#include "utils.h"
// Global:
char *__logdir;
int __loginterval;
int __port;
void publish(void *publisher, char *filepath, char* client, int interval);
int main (int argc, char *argv [])
{
if(argc != 4){
printf("Usage: %s <abslogdir> <loginterval> <port>",argv[0]);
exit(1);
}
//----- Init global variables
__logdir=argv[1];
__loginterval=atoi(argv[2]);
__port=atoi(argv[3]);
//----- Prepare our context and publisher
void *context = zmq_ctx_new ();
void *publisher = zmq_socket (context, ZMQ_PUB);
char bindto[30];
sprintf(bindto,"tcp://*:%d",__port);
int rc = zmq_bind (publisher, bindto);
if(rc!=0){
printf("Failed to bind zmq on %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);
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...%s\n",client,logfile_next);
while(!FILE_EXISTS(logfile_next)){
sleep(1);
}
// Send current one
if(FILE_EXISTS(logfile)){
publish(publisher,logfile,client,interval);
remove(logfile);
}
}
}
closedir(dr);
}
zmq_close (publisher);
zmq_ctx_destroy (context);
return 0;
// // Prepare our context and publisher
// void *context = zmq_ctx_new ();
// void *publisher = zmq_socket (context, ZMQ_PUB);
// int rc = zmq_bind (publisher, "tcp://*:"STRINGIFY(PUBLISHER_PORT));
// assert (rc == 0);
// // Initialize random number generator
// while (1) {
// zmq_send (publisher, "Hello World", 5, 0);
// printf("AA\n");
// }
// zmq_close (publisher);
// zmq_ctx_destroy (context);
return 0;
}
void publish(void *publisher, char *filepath, char* client, int interval){
printf("Publish!\n");
zmq_send (publisher, ZMQ_TOKEN, strlen(ZMQ_TOKEN), 0);
}
|
