loic/ina260-zmq-publisher
1
0
Fork 0
mirror of https://gitlab.com/manzerbredes/ina260-zmq-publisher.git synced 2025-04-19 12:19:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Minor changes

Browse source
This commit is contained in:
Loic Guegan 2023-07-17 16:41:07 +02:00
parent 0c08c2f0d8
commit c79a02c57b
4 changed files with 31 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

49
src/logger.c
View file

@ -20,8 +20,6 @@ char __logdir[STATIC_LEN];
char __regpower[STATIC_LEN];
int __loginterval;
unsigned char __stop=0;
void *__zmq_context;
void *__zmq_publisher;
void sighandler(int signo){
if (signo == SIGINT){
@ -30,7 +28,7 @@ void sighandler(int signo){
}
}
void publish(int queue_id);
void publish(int queue_id, void* publisher);
typedef struct record {
time_t secs;
@ -77,21 +75,21 @@ 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);
}
//----- Prepare our context and publisher
__zmq_context = zmq_ctx_new ();
__zmq_publisher = zmq_socket (__zmq_context, ZMQ_PUB);
char bindto[STATIC_LEN];
sprintf(bindto,"tcp://%s:%d",__ip,__port);
int rc = zmq_connect (__zmq_publisher, bindto);
if(rc!=0){
printf("Failed to connect to %s\n",bindto);
exit(1);
}
void *context = zmq_ctx_new ();
void *publisher = zmq_socket (context, ZMQ_PUB);
char bindto[STATIC_LEN];
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);
}
@ -100,7 +98,7 @@ int main (int argc, char *argv [])
FILE *regptr,*logptr;
char logfilepath[STATIC_LEN]="";
regptr=fopen("/home/loic/out.txt","r");
regptr=fopen(__regpower,"r");
char buffer[STATIC_LEN];
int power;
time_t interval;
@ -110,7 +108,8 @@ int main (int argc, char *argv [])
while(!__stop){
interval=INTERVAL(__loginterval);
// Log current interval
queue_id=queue_id>=RECORD_QUEUES ? 0 : queue_id+1;
queue_id=(queue_id+1)>=RECORD_QUEUES ? 0 : (queue_id+1);
printf("Queue id %d",queue_id);
queues[queue_id].interval=interval;
int record=0;
while((TIMESTAMP()-interval)<__loginterval){
@ -136,41 +135,43 @@ int main (int argc, char *argv [])
}
}
queues[queue_id].length=record;
publish(queue_id);
publish(queue_id,publisher);
}
fclose(regptr);
zmq_close (__zmq_publisher);
zmq_ctx_destroy (__zmq_context);
zmq_close (publisher);
zmq_ctx_destroy (context);
return 0;
}
void publish(int queue_id){
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);
printf(buffer);
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",queues[queue_id].records[record].secs,queues[queue_id].records[record].nsecs,queues[queue_id].records[record].power);
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){
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 (__zmq_publisher, buffer, msglen, 0);
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 (__zmq_publisher, buffer, msglen, 0);
zmq_send (publisher, buffer, msglen, 0);
}