aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/broevent.c85
-rw-r--r--src/fifoqueue.c132
-rw-r--r--src/midbro.c74
3 files changed, 291 insertions, 0 deletions
diff --git a/src/broevent.c b/src/broevent.c
new file mode 100644
index 0000000..362ed5e
--- /dev/null
+++ b/src/broevent.c
@@ -0,0 +1,85 @@
+#include "fifoqueue.h"
+#include "broevent.h"
+#ifdef BROCCOLI
+#include <broccoli.h>
+#endif
+
+char *host_default = "127.0.0.1";
+char *port_default = "47760";
+Fifo_q * q;
+
+ static void
+modbus_register_received(BroConn *conn, void *data, BroRecord *record)
+{
+ int type = BRO_TYPE_COUNT;
+ uint64 *address = NULL;
+ uint64 *value = NULL;
+
+ // TODO: handle regtype
+ address = bro_record_get_named_val(record, "address", &type);
+ if (!address) {
+ // TODO: handle error
+ return;
+ }
+ value = bro_record_get_named_val(record, "register", &type);
+ if (!value) {
+ // TODO: handle error
+ return;
+ }
+ #ifdef DEBUG
+ printf("Received value %"PRIu64" from uid=%"PRIu64"\n",*value,*address);
+ #endif
+
+ add_to_queue(q, create_sensor_object(*value, *address));
+
+ #ifdef DEBUG
+ printf("Added to queue.\n");
+ #endif
+}
+
+ void *
+bro_event_listener(void * args)
+{
+ q = (Fifo_q *) args;
+ int fd = -1;
+ BroConn *bc = NULL;
+ bro_init(NULL);
+ char hostname[512];
+
+ snprintf(hostname, 512, "%s:%s", host_default, port_default);
+ if (! (bc = bro_conn_new_str(hostname, BRO_CFLAG_RECONNECT | BRO_CFLAG_ALWAYS_QUEUE)))
+ {
+ printf("Could not get Bro connection handle.\n");
+ exit(-1);
+ }
+ bro_debug_calltrace = 0;
+ bro_debug_messages = 0;
+
+ bro_event_registry_add(bc, "pasad_register_received",
+ (BroEventFunc) modbus_register_received, NULL);
+
+ if (! bro_conn_connect(bc))
+ {
+ printf("Could not connect to Bro at %s:%s.\n", host_default,
+ port_default);
+ exit(-1);
+ }
+
+ fd =bro_conn_get_fd(bc);
+ fd_set rfds;
+ setbuf(stdout,NULL);
+
+ while(true)
+ {
+ FD_ZERO(&rfds);
+ FD_SET(fd,&rfds);
+ if(select(fd+1,&rfds,NULL,NULL,NULL) == -1){
+ printf("select(): Bad file descriptor");
+ break;
+ }
+
+ bro_conn_process_input(bc);
+ }
+
+ bro_conn_delete(bc);
+}
diff --git a/src/fifoqueue.c b/src/fifoqueue.c
new file mode 100644
index 0000000..e7ed8f6
--- /dev/null
+++ b/src/fifoqueue.c
@@ -0,0 +1,132 @@
+#include <pthread.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <semaphore.h>
+#include "types.h"
+#include "fifoqueue.h"
+
+ Fifo_q *
+init_queue(int size)
+{
+ Fifo_q * q = (Fifo_q *) malloc(sizeof(Fifo_q));
+ q->head = NULL;
+ q->tail = NULL;
+ q->maxSize = size;
+ q->currentSize = 0;
+ q->droppedValues = 0;
+ q->largestBufferSize = 0;
+ q->valuesReceived = 0;
+ q->valuesReleased = 0;
+ /*Queue empty from the beginning (block)*/
+ sem_init(&q->bufferEmptyBlock, 0, 0);
+ sem_init(&q->bufferFullBlock, 0, size);
+ sem_init(&q->lock, 0, 1);
+ return q;
+}
+
+ boolean
+is_full(Fifo_q * q)
+{
+ if(q->currentSize < q->maxSize)
+ return false;
+ else
+ return true;
+}
+
+ boolean
+is_empty(Fifo_q * q)
+{
+ if(q->head==NULL)
+ return true;
+ else
+ return false;
+}
+
+ int
+add_to_queue(Fifo_q * q, Sensor_t * sensor)
+{
+
+ if(q == NULL){
+ printf("Error: Queue not initialized\n");
+ free(sensor); //free if not appended
+ return -1;
+ }
+ /* Drop Least Recently or Drop Most Recently */
+ #ifdef DLR
+ if(is_full(q)){
+ pop_from_queue(q);
+ q->droppedValues++;
+ return 0;
+ }
+ #else
+ sem_wait(&q->bufferFullBlock);
+ #endif
+ sem_wait(&q->lock);
+ Queue_t * new_elem = (Queue_t *) malloc(sizeof(Queue_t));
+ new_elem->next = NULL;
+ new_elem->sensor = sensor;
+ if(is_empty(q)){
+ q->head = new_elem;
+ }else
+ q->tail->next = new_elem;
+ q->tail = new_elem;
+ q->currentSize++;
+ q->valuesReceived++;
+ if(q->currentSize > q->largestBufferSize)
+ q->largestBufferSize = q->currentSize;
+ sem_post(&q->lock);
+ sem_post(&q->bufferEmptyBlock);
+ return 1;
+}
+
+ Sensor_t *
+pop_from_queue(Fifo_q * q)
+{
+ int semStat;
+ sem_wait(&q->bufferEmptyBlock);
+ sem_wait(&q->lock);
+ Queue_t * head = q->head;
+ Sensor_t * sensor = head->sensor;
+ /* If dequeue the last element */
+ if(q->currentSize == 1){
+ q->head = NULL;
+ q->tail = NULL;
+ }else{
+ q->head = head->next;
+ }
+ free(head);
+ q->currentSize--;
+ q->valuesReleased++;
+ sem_post(&q->lock);
+ #ifndef DLR
+ sem_post(&q->bufferFullBlock);
+ #endif
+ 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)
+{
+ sem_wait(&q->lock);
+ Queue_t * current = q->head;
+ printf("\nContent of the queue with size=%d\n",q->currentSize);
+ if(current == NULL){
+ printf("The queue is empty!\n");
+ sem_post(&q->lock);
+ return;
+ }
+ while(current != NULL){
+ printf("sensor value=%d, sensor uid=%d\n",
+ current->sensor->value, current->sensor->uid);
+ current = current->next;
+ }
+ sem_post(&q->lock);
+}
diff --git a/src/midbro.c b/src/midbro.c
new file mode 100644
index 0000000..e2f9644
--- /dev/null
+++ b/src/midbro.c
@@ -0,0 +1,74 @@
+#include <pthread.h>
+#include <unistd.h>
+#include <signal.h>
+#include "fifoqueue.h"
+#include "broevent.h"
+#include "midbro.h"
+#ifdef BROCCOLI
+#include <broccoli.h>
+#endif
+
+Fifo_q * queue;
+pthread_t event_listener;
+sigset_t signal_set;
+
+ void
+sigint_handler(int signal)
+{
+ printf("\nStatistics:\n"
+ "Total values received: %d\n"
+ "Total values dropped: %d\n"
+ "Total values released: %d\n"
+ "Maximum buffer utilization: %d\n"
+ "Buffer fixed size: %d\n"
+ "Buffer size upon termination: %d\n",
+ queue->valuesReceived, queue->droppedValues,
+ queue->valuesReleased, queue->largestBufferSize,
+ queue->maxSize, queue->currentSize);
+ exit(0);
+}
+ void
+request_n_values(int number, int arrayOfValues[])
+{
+ int i;
+ Sensor_t * sensor;
+ for(i=0; i<number; ++i){
+ sensor = pop_from_queue(queue);
+ arrayOfValues[i] = sensor->value;
+ free(sensor);
+ }
+ printf("Release %d sensor data values\n", number);
+}
+ int
+request_value()
+{
+ int value;
+ Sensor_t * sensor;
+ sensor = pop_from_queue(queue);
+ value = sensor->value;
+ free(sensor);
+ printf("Release 1 sensor data value\n");
+ return value;
+}
+
+ void
+start_data_capture()
+{
+ int res;
+ queue = init_queue(500); /* Initiate queue with fixed size */
+ /* Create producer thread that listen for bro events */
+ sigemptyset(&signal_set);
+ sigaddset(&signal_set, SIGINT);
+ res = pthread_sigmask(SIG_BLOCK, &signal_set, NULL);
+ if(res != 0)
+ perror("SIGINT block");
+ res = pthread_create(&event_listener, NULL, bro_event_listener, queue);
+ if(res){
+ perror("Unable to create thread");
+ exit(-1);
+ }
+ res = pthread_sigmask(SIG_UNBLOCK, &signal_set, NULL);
+ if(res != 0)
+ perror("SIGINT unblock");
+ signal(SIGINT, sigint_handler);
+}
generated by cgit v1.2.3 (git 2.46.0) at 2024-12-23 09:04:10 +0000