graceful stop of sink

commonTypes.h

@@ -9,7 +9,8 @@
 typedef struct {
     config_t *config;
     ringbuffer_t *ringbuffer;
-    void* context;
+    void* mqttContext;
+    void* sinkContext;
 } commonThreadHandle_t;
 
 typedef struct {

mqttauditing.c

@@ -66,6 +66,11 @@ int main (void) {
   pthread_join(mqttThread, NULL);
   fprintf(stderr, "mqtt receiver thread joined\n");
 
+  sinkStop((void*) &commonThreadHandle);
+  ringbufferInterrupt(&ringbuffer);
+  pthread_join(sinkThread, NULL);
+  fprintf(stderr, "sink thread joined\n");
+
 
 
   // will never be reached

mqttreceiver.c

@@ -49,7 +49,7 @@ typedef struct {
 
 int on_message(void *kontext, char *topicName, int topicLen, MQTTClient_message *message) {
     commonThreadHandle_t *handle = (commonThreadHandle_t*)kontext;
-    mqttThreadContext_t *context = (mqttThreadContext_t*)handle->context;
+    mqttThreadContext_t *context = (mqttThreadContext_t*)handle->mqttContext;
 
     char* payload = message->payload;
     printf("Topic: %s, TopicLen: %d, Payload: %s\n", topicName, topicLen, payload);
@@ -83,7 +83,7 @@ int on_message(void *kontext, char *topicName, int topicLen, MQTTClient_message
 
 void on_disconnect(void *kontext, char *cause) {
     commonThreadHandle_t *handle = (commonThreadHandle_t*)kontext;
-    mqttThreadContext_t *context = (mqttThreadContext_t*)handle->context;
+    mqttThreadContext_t *context = (mqttThreadContext_t*)handle->mqttContext;
 
     fprintf(stderr, "disconnected from broker: %s trying to reconnect\n", cause);
 
@@ -121,7 +121,7 @@ void *mqttreceiver(void *ptr) {
 
     commonThreadHandle_t *handle = (commonThreadHandle_t*)ptr;
     mqttThreadContext_t *context = (mqttThreadContext_t*) malloc(sizeof(mqttThreadContext_t));
-    handle->context = (void*)context;
+    handle->mqttContext = (void*)context;
     context->stopSignal = false;
     context->watchdogCounter = 0;
 

ringbuffer.c

@@ -8,11 +8,20 @@
 void ringbufferInit(ringbuffer_t *handle) {
     handle->bufferReadIdx = 0;
     handle->bufferWriteIdx = 0;
+    handle->interrupted = false;
     pthread_mutex_init(&(handle->eventMutex), NULL);
     pthread_cond_init(&(handle->eventSignal), NULL);
     fprintf(stderr, "ringbuffer initialized\n");
 }
 
+void ringbufferInterrupt(ringbuffer_t *handle) {
+    handle->interrupted = true;
+
+    pthread_mutex_lock(&(handle->eventMutex));
+    pthread_cond_signal(&(handle->eventSignal));
+    pthread_mutex_unlock(&(handle->eventMutex));
+}
+
 void ringbufferPut(ringbuffer_t *handle, void *f) {
     if (handle->bufferWriteIdx == (BUFFER_SIZE - 1)) {
         while (handle->bufferReadIdx == BUFFER_SIZE);
@@ -34,16 +43,22 @@ void ringbufferPut(ringbuffer_t *handle, void *f) {
 
 
 void *ringbufferGet(ringbuffer_t *handle) {
+    void *res = NULL;
+
     if (handle->bufferReadIdx == handle->bufferWriteIdx) {
         pthread_mutex_lock(&(handle->eventMutex));
         pthread_cond_wait(&(handle->eventSignal), &(handle->eventMutex));
         pthread_mutex_unlock(&(handle->eventMutex));
     }
 
-    void *res = handle->buffer[handle->bufferReadIdx];
-    handle->bufferReadIdx++;
-    if (handle->bufferReadIdx > BUFFER_SIZE) {
-        handle->bufferReadIdx = 0;
+    if (handle->interrupted) {
+        handle->interrupted = false;
+    } else {       
+        res = handle->buffer[handle->bufferReadIdx];
+        handle->bufferReadIdx++;
+        if (handle->bufferReadIdx > BUFFER_SIZE) {
+           handle->bufferReadIdx = 0;
+        }
     }
     return res;
 }

ringbuffer.h

@@ -3,7 +3,7 @@
 
 #include <stdint.h>
 #include <pthread.h>
-
+#include <stdbool.h>
 
 #define BUFFER_SIZE 256
 
@@ -11,6 +11,7 @@ typedef struct {
     void* buffer[BUFFER_SIZE+5];
     uint32_t bufferReadIdx;
     uint32_t bufferWriteIdx;
+    bool interrupted;
     pthread_mutex_t eventMutex;
     pthread_cond_t eventSignal;
 } ringbuffer_t;
@@ -18,5 +19,6 @@ typedef struct {
 void ringbufferInit(ringbuffer_t *handle);
 void ringbufferPut(ringbuffer_t *handle, void *f);
 void *ringbufferGet(ringbuffer_t *handle);
+void ringbufferInterrupt(ringbuffer_t *handle);
 
-#endif // _RINGBUFFER_H_
+#endif // _RINGBUFFER_H_

sink.c

@@ -1,27 +1,51 @@
 #include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <stdbool.h>
 #include "commonTypes.h"
 #include "ringbuffer.h"
 
+
+
+typedef struct {
+    bool stopSignal;
+} sinkThreadContext_t;
+
+void sinkStop(void *ptr) {
+    commonThreadHandle_t *handle = (commonThreadHandle_t*)ptr;
+    sinkThreadContext_t *context = (sinkThreadContext_t*)handle->sinkContext;
+
+    context->stopSignal = true;
+    fprintf(stderr, "sink thread, stop flagged\n");
+}
+
+
 void *sink(void *ptr) {
     fprintf(stderr, "sink entered\n");
 
     commonThreadHandle_t *handle = (commonThreadHandle_t*)ptr;
+    sinkThreadContext_t *context = (sinkThreadContext_t*) malloc(sizeof(sinkThreadContext_t));
+    handle->sinkContext = (void*)context;
+    context->stopSignal = false;
 
-    while (1) {
+    while (! context->stopSignal) {
         auditItem_t *auditItem = (auditItem_t*)ringbufferGet(handle->ringbuffer);
 
-        printf("AuditItem: Time:    %ld\n", auditItem->ts);
-        printf("AuditItem: Topic:   %s\n", auditItem->topic);
-        printf("AuditItem: Payload: %s\n", auditItem->payload);
+        if (auditItem != NULL) {
+            printf("AuditItem: Time:    %ld\n", auditItem->ts);
+            printf("AuditItem: Topic:   %s\n", auditItem->topic);
+            printf("AuditItem: Payload: %s\n", auditItem->payload);
 
-        free(auditItem->topic);
-        auditItem->topic = NULL;
-        free(auditItem->payload);
-        auditItem->payload = NULL;
-        free(auditItem);
-        auditItem = NULL;
+            free(auditItem->topic);
+            auditItem->topic = NULL;
+            free(auditItem->payload);
+            auditItem->payload = NULL;
+            free(auditItem);
+            auditItem = NULL;
+        }
     }
+
+    fprintf(stderr, "sink thread stopped\n");
+    return (void*)NULL;
 }
 

sink.h

@@ -2,5 +2,7 @@
 #define _SINK_H_
 
 void *sink(void *ptr);
+void sinkStop(void *ptr);
+
 
 #endif // _SINK_H_