Minor changes

cdata/node1/1234/1689609135

@@ -0,0 +1 @@
+1689609135,618497985,56871689609136,618687904,56871689609137,618845801,56871689609138,619042679,56871689609139,619229375,5687

cdata/node1/1234/1689609140

@@ -0,0 +1 @@
+1689609140,619495269,56871689609141,619673980,56871689609142,619870550,56871689609143,620054237,56871689609144,620245991,5687

cdata/node1/1234/1689609145

@@ -0,0 +1 @@
+1689609145,620461509,56871689609146,620618837,56871689609147,620819124,5687

cdata/node1/1234/1689609180

@@ -0,0 +1 @@
+1689609184,738852819,5687

cdata/node1/1234/1689609185

@@ -0,0 +1 @@
+1689609185,739106591,56871689609186,739282589,56871689609187,739481520,56871689609188,739603953,56871689609189,739779924,5687

cdata/node1/1234/1689609215

@@ -0,0 +1,3 @@
+1689609217,274459175,5687
+1689609218,274690512,5687
+1689609219,274792657,5687

cdata/node1/1234/1689609220

@@ -0,0 +1,5 @@
+1689609220,275031326,5687
+1689609221,275119233,5687
+1689609222,275298864,5687
+1689609223,275614796,5687
+1689609224,275750180,5687

cdata/node1/1234/1689609225

@@ -0,0 +1,5 @@
+1689609225,276040681,5687
+1689609226,276221040,5687
+1689609227,276406843,5687
+1689609228,276601805,5687
+1689609229,276787994,5687

cdata/node1/1234/1689609230

@@ -0,0 +1,5 @@
+1689609230,277018992,5687
+1689609231,277135987,5687
+1689609232,277323469,5687
+1689609233,277506592,5687
+1689609234,277692437,5687

cdata/node1/1234/1689609235

@@ -0,0 +1,2 @@
+1689609235,277925943,5687
+1689609236,278107452,5687

cdata/node1/1234/1689609485

@@ -0,0 +1,4 @@
+1689609486,977646780,5687
+1689609487,977839219,5687
+1689609488,978023233,5687
+1689609489,978203588,5687

cdata/node1/1234/1689609490

@@ -0,0 +1,5 @@
+1689609490,978369193,5687
+1689609491,978487394,5687
+1689609492,978672875,5687
+1689609493,978858663,5687
+1689609494,979046614,5687

cdata/node1/1234/1689609495

@@ -0,0 +1,2 @@
+1689609495,979264184,5687
+1689609496,979355744,5687

config.mk

@@ -19,7 +19,7 @@ LOGGERS_DIR=/tmp/ina260_logs/
 # LOGGERS_DELAY defines the delay between 2 consecutive
 # ina260 power read performed by the logger
 # Unit is milliseconds
-LOGGERS_DELAY=0
+LOGGERS_DELAY=1000
 # SUBSCRIBER_DIR will contain all the measurments
 # received from the publishers
 SUBSCRIBER_DIR=./data

src/logger.c

@@ -30,19 +30,12 @@ void sighandler(int signo){
 
 void publish(int queue_id, void* publisher);
 
-typedef struct record {
-  time_t secs;
-  long nsecs;
-  uint16_t power;
-} record;
-
 typedef struct queue {
-  int length;
-  long interval;
-  record records[RECORD_MAX];
+  int size;
+  char msg[ZMQ_MSG_SIZE];
 } queue;
 
-queue queues[RECORD_QUEUES];
+queue queues[MAX_QUEUES];
 
 int main (int argc, char *argv [])
 {
@@ -75,7 +68,7 @@ int main (int argc, char *argv [])
     printf("Log interval is too small (min=%ds)\n",MIN_INTERVAL);
     exit(2);
   }
-  if(!FILE_EXISTS(__regpower)){
+  if(FILE_EXISTS(__regpower)){
     printf("Logger cannot access to %s\n",__regpower);
     exit(3);
   }
@@ -98,43 +91,45 @@ int main (int argc, char *argv [])
 
   FILE *regptr,*logptr;
   char logfilepath[STATIC_LEN]="";
-  regptr=fopen(__regpower,"r");
+  regptr=fopen("/home/loic/out.txt","r");
   char buffer[STATIC_LEN];
   int power;
   time_t interval;
   struct timespec power_ts;
   int queue_id=0;
+  char *header[STATIC_LEN];
 
   while(!__stop){
     interval=INTERVAL(__loginterval);
     // Log current interval
-    queue_id=(queue_id+1)>=RECORD_QUEUES ? 0 : (queue_id+1);
+    queue_id=(queue_id+1)>=MAX_QUEUES ? 0 : (queue_id+1);
     printf("Queue id %d",queue_id);
-    queues[queue_id].interval=interval;
-    int record=0;
+    *queues[queue_id].msg='\0';
+    sprintf(queues[queue_id].msg,"%s\n%s\n%s\n%ld\n",ZMQ_TOKEN,__key,__client,interval);
+    queues[queue_id].size=strlen(queues[queue_id].msg);
     while((TIMESTAMP()-interval)<__loginterval){
       if(__stop)
         break;
-      if(record < RECORD_MAX){
-        // Read power:
-        fgets(buffer,STATIC_LEN,regptr);
-        // Get power measurement timestamp:
-        clock_gettime(CLOCK_REALTIME,&power_ts);
-        queues[queue_id].records[record].secs=power_ts.tv_sec;
-        queues[queue_id].records[record].nsecs=power_ts.tv_nsec;
-        queues[queue_id].records[record].power=atoi(buffer);
-        // Reset power register file:
-        fseek(regptr,0,SEEK_SET);
-  #if LOGGERS_DELAY > 0
-        usleep(LOGGERS_DELAY*1000);
-  #endif
-        printf("Tick\n"); fflush(stdout);
-        record++;
+      // Read power:
+      fgets(buffer,STATIC_LEN,regptr);
+      // Get power measurement timestamp:
+      clock_gettime(CLOCK_REALTIME,&power_ts);
+      char line[STATIC_LEN];
+      sprintf(line,"%ld,%ld,%d\n",power_ts.tv_sec,power_ts.tv_nsec,atoi(buffer));
+      int linelen=strlen(line);
+      if(queues[queue_id].size+linelen>ZMQ_MSG_SIZE){
+        printf("To many measurements to publish. Please increase ZMQ_MSG_SIZE\n");
       } else {
-        printf("Queue overflow, RECORD_MAX must be increase!! n=%d\n",record);
+        strcat(queues[queue_id].msg,line);
+        queues[queue_id].size+=strlen(line);
       }
+      // Reset power register file:
+      fseek(regptr,0,SEEK_SET);
+#if LOGGERS_DELAY > 0
+      usleep(LOGGERS_DELAY*1000);
+#endif
+      printf("Tick\n"); fflush(stdout);
     }
-    queues[queue_id].length=record;
     publish(queue_id,publisher);
   }
 
@@ -148,30 +143,5 @@ int main (int argc, char *argv [])
 
 void publish(int queue_id, void* publisher){
     printf("Publishing...\n");
-    // Build message header:
-    char buffer[ZMQ_MSG_SIZE];
-    sprintf(buffer,"%s\n%s\n%s\n%ld\n",ZMQ_TOKEN,__key,__client,queues[queue_id].interval);
-    int msglen=strlen(buffer);
-
-    // Put every lines in the buffer and send it
-    char line[STATIC_LEN];
-    for(int record=0;record<queues[queue_id].length;record++){
-      *line='\0';
-      sprintf(line,"%ld,%ld,%d\n",queues[queue_id].records[record].secs,queues[queue_id].records[record].nsecs,queues[queue_id].records[record].power);
-      int linelength=strlen(line);
-      if((linelength+msglen)<ZMQ_MSG_SIZE){
-          strcat(buffer,line);
-          msglen+=linelength;
-      } else {
-          // It buffer is full, we send the message and create another one
-          zmq_send (publisher, buffer, msglen, 0);
-          // Build new message header:
-          sprintf(buffer,"%s\n%s\n%s\n%ld\n",ZMQ_TOKEN,__key,__client,queues[queue_id].interval);
-          strcat(buffer,line);
-          msglen=strlen(buffer);
-      }
-    }
-    printf("msglength=%d\n",msglen);
-    // Finally send the last message (or the only one)
-    zmq_send (publisher, buffer, msglen, 0);
+    zmq_send(publisher,queues[queue_id].msg,queues[queue_id].size,0);
 }

src/subscriber.c

@@ -79,6 +79,7 @@ int main (int argc, char *argv [])
       // Write all the measurements:
       if(line>4){
         fwrite(token, strlen(token), 1, fptr);
+        printf("%s\n",token);
         fwrite("\n",1,1,fptr);
       }
 

src/utils.h

@@ -10,8 +10,7 @@
 #define INA260_SYSFS "/sys/kernel/ina260"
 #define INA260_POWER_REGISTER "registers/power"
 #define STATIC_LEN 255
-#define RECORD_QUEUES 1
-#define RECORD_MAX 100000
+#define MAX_QUEUES 1
 
 #ifndef ZMQ_TOKEN
 #define ZMQ_TOKEN "ina260-zmq-publisher"