Minor changes

1 files changed, 80 insertions, 0 deletions

diff --git a/src/subscriber.c b/src/subscriber.c
new file mode 100644
index 0000000..932a82d
--- /dev/null
+++ b/src/subscriber.c
@@ -0,0 +1,80 @@
+#include <zmq.h>
+#include <assert.h>
+#include <time.h>
+#include <string.h>
+
+#include "utils.h"
+
+int main (int argc, char *argv [])
+{
+  if(argc != 3){
+    printf("Usage: %s <port> <cdatadir>",argv[0]);
+    exit(1);
+  }
+
+  //----- Arguments
+  int port=atoi(argv[1]);
+  char *cdatadir=argv[2];
+
+  //----- Various inits
+  mkdirp(cdatadir);
+
+  //----- Init ZMQ
+  void *context = zmq_ctx_new ();
+  void *subscriber = zmq_socket (context, ZMQ_SUB);
+  char bindto[30];
+  sprintf(bindto,"tcp://*:%d",port);
+  int rc = zmq_bind (subscriber, bindto);
+  if(rc!=0){
+    printf("Failed to bind zmq on %s\n",bindto);
+    exit(1);
+  }
+  rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,
+                        ZMQ_TOKEN, strlen(ZMQ_TOKEN));
+
+  //----- Listen
+  char buffer[ZMQ_MSG_SIZE];
+  int size;
+  while(1){
+    size=zmq_recv (subscriber, buffer, ZMQ_MSG_SIZE-1, 0);
+    buffer[size < ZMQ_MSG_SIZE ? size : ZMQ_MSG_SIZE - 1] = '\0';
+    //----- Read buffer
+    char *token = strtok(buffer, "\n");
+    char key[255];
+    char client[255];
+    long int interval;
+    FILE *fptr;
+    int line=1;
+    while(token != NULL){
+      if(line==2)
+        strcpy(key,token);
+      else if(line==3)
+        strcpy(client,token);
+      else if(line==4)
+        interval=atoi(token);
+
+      if(line==4){
+        printf("Data received with key=%s\n",key);
+        char path[255]="";
+        sprintf(path,"%s/%s_%s_%ld",cdatadir,key,client,interval);
+        fptr=fopen(path,"a");
+      }
+
+      if(line>4){
+        fwrite(token, strlen(token), 1, fptr);
+        fwrite("\n",2,1,fptr);
+      }
+
+      token=strtok(NULL, "\n");
+      line++;
+    }
+    fclose(fptr);
+  }
+
+
+
+  zmq_close (subscriber);
+  zmq_ctx_destroy (context);
+
+  return 0;
+}
\ No newline at end of file