Add new thread for event handling

6 files changed, 52 insertions, 9 deletions

diff --git a/broccoli/Makefile b/broccoli/Makefile
index 3fafbea..d6e2a37 100644
--- a/broccoli/Makefile
+++ b/broccoli/Makefile
@@ -1,6 +1,6 @@
 CC=gcc
 CFLAGS = -c -Wall -I/usr/local/include -I/usr/local/include -DBROCCOLI 
-LDFLAGS =  -L/usr/local/lib -lbroccoli
+LDFLAGS =  -L/usr/local/lib -lbroccoli -pthread
 
 all: midbropasad
 
diff --git a/broccoli/broevent.c b/broccoli/broevent.c
index 94d06b8..2b3db8d 100644
--- a/broccoli/broevent.c
+++ b/broccoli/broevent.c
@@ -3,20 +3,22 @@
 
 char *host_default = "127.0.0.1";
 char *port_default = "47760";
+Fifo_q * q;
 
     static void
 bro_response(BroConn *conn, void *data, uint64* registers, uint64* uid)
 {
-    printf("Received value %"PRIu64" from uid=%"PRIu64"\n",*registers,*uid);
+    add_to_queue(q,create_sensor_object(*registers,*uid));
+    //printf("Received value %"PRIu64" from uid=%"PRIu64"\n",*registers,*uid);
 
     conn = NULL;
     data = NULL;
 }
 
-    void
-bro_event_listener()
+    void *
+bro_event_listener(void * args)
 {
-
+    q = (Fifo_q *) args;
     int fd = -1;
     BroConn *bc = NULL;
     bro_init(NULL);
diff --git a/broccoli/fifoqueue.c b/broccoli/fifoqueue.c
index 56b1dda..6676484 100644
--- a/broccoli/fifoqueue.c
+++ b/broccoli/fifoqueue.c
@@ -1,4 +1,7 @@
 #include "includes/fifoqueue.h"
+#include <pthread.h>
+
+pthread_mutex_t lock;
 
     Fifo_q * 
 init_queue(int size)
@@ -8,6 +11,11 @@ init_queue(int size)
     q->tail = NULL;
     q->maxSize = size;
     q->currentSize = 0;
+
+    if (pthread_mutex_init(&lock, NULL) != 0)
+    {
+        printf("WARNING: Couldn't initialize lock\n");
+    }
     return q;
 }
 
@@ -32,6 +40,8 @@ is_empty(Fifo_q * q)
     int
 add_to_queue(Fifo_q * q, Sensor_t * sensor)
 {
+
+    pthread_mutex_lock(&lock);
     /* TODO delete first one if full */
     if(q == NULL){
         return -1;    
@@ -48,12 +58,16 @@ add_to_queue(Fifo_q * q, Sensor_t * sensor)
         q->tail->next = new_elem;
     q->tail = new_elem;
     q->currentSize++;
+    pthread_mutex_unlock(&lock);
+    print_queue(q);
     return 1;
 }
 
     Sensor_t *
 pop_from_queue(Fifo_q * q)
 {
+
+    pthread_mutex_lock(&lock);
     if(is_empty(q)){
         perror("The queue is empty");
         exit(-1);
@@ -63,12 +77,21 @@ pop_from_queue(Fifo_q * q)
     Sensor_t * sensor = head->sensor;
     free(head);
     q->currentSize--;
+    pthread_mutex_unlock(&lock);
     return sensor;
 } 
 
+    Sensor_t *
+create_sensor_object(int value, int uid){
+    Sensor_t * sensor = (Sensor_t *) malloc(sizeof(Sensor_t *));
+    sensor->value = value;
+    sensor->uid = uid;
+    return sensor;
+}
     void
 print_queue(Fifo_q * q)
 {
+    pthread_mutex_lock(&lock);
     Queue_t * current = q->head;
     if(current == NULL){
         printf("The queue is empty!");
@@ -79,4 +102,5 @@ print_queue(Fifo_q * q)
                 current->sensor->value, current->sensor->uid);
         current = current->next;
     }
+    pthread_mutex_unlock(&lock);
 }
diff --git a/broccoli/includes/broevent.h b/broccoli/includes/broevent.h
index 4021737..653796e 100644
--- a/broccoli/includes/broevent.h
+++ b/broccoli/includes/broevent.h
@@ -2,4 +2,4 @@
 #include <broccoli.h>
 #endif
 
-void bro_event_listener();
+void * bro_event_listener(void * args);
diff --git a/broccoli/includes/fifoqueue.h b/broccoli/includes/fifoqueue.h
index 9ee5c53..398e1e9 100644
--- a/broccoli/includes/fifoqueue.h
+++ b/broccoli/includes/fifoqueue.h
@@ -4,6 +4,8 @@
 
 Fifo_q * init_queue(int size);
 
+Sensor_t * create_sensor_object(int value, int uid);
+
 boolean is_full(Fifo_q * q);
 
 boolean is_empty(Fifo_q * q);
diff --git a/broccoli/midbropasad.c b/broccoli/midbropasad.c
index 5c0c413..6db59a5 100644
--- a/broccoli/midbropasad.c
+++ b/broccoli/midbropasad.c
@@ -1,12 +1,27 @@
 #include "includes/fifoqueue.h"
 #include "includes/broevent.h"
 
+    void
+start_data_capture(Fifo_q * q)
+{
+    int res;
+    pthread_t event_listener;
+    res = pthread_create(&event_listener, NULL, bro_event_listener, q);
+    if(res){
+        perror("Unable to create thread");
+        exit(-1);
+    }
+}
+
     int
 main(int argc, char **argv)
 {
-    Fifo_q * q = init_queue(5);
-    bro_event_listener();
-
+    Fifo_q * q = init_queue(50);
+    start_data_capture(q);
+    while(true){
+        printf("Main thread\n");
+        sleep(2);
+    }
     free(q);
     return 0;
 }