dispatcher.c

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <uv.h>

typedef void (*event_handler)(void *);

typedef struct {
  uv_timer_t timer_req;
  event_handler handler;
  void *data;
  unsigned long timeout;
} timer_event;

typedef struct queue_node {
  void *data;
  struct queue_node *next;
} queue_node;

typedef struct {
  queue_node *head, *tail;
  pthread_mutex_t lock;
} queue;

void queue_init(queue *q) {
  q->head = q->tail = NULL;
  pthread_mutex_init(&q->lock, NULL);
}

void queue_push(queue *q, void *data) {
  queue_node *new_node = malloc(sizeof(queue_node));
  new_node->data = data;
  new_node->next = NULL;

  pthread_mutex_lock(&q->lock);
  if (!q->tail) {
    q->head = q->tail = new_node;
  } else {
    q->tail->next = new_node;
    q->tail = new_node;
  }
  pthread_mutex_unlock(&q->lock);
}

void *queue_pop(queue *q) {
  pthread_mutex_lock(&q->lock);
  if (!q->head) {
    pthread_mutex_unlock(&q->lock);
    return NULL;
  }

  queue_node *node = q->head;
  void *data = node->data;
  q->head = q->head->next;
  if (!q->head) {
    q->tail = NULL;
  }
  pthread_mutex_unlock(&q->lock);

  free(node);
  return data;
}

typedef struct {
  uv_loop_t *loop;
  uv_async_t async_req;
  queue event_queue;
} dispatcher;

void timer_callback(uv_timer_t *handle) {
  timer_event *event = (timer_event *)handle->data;
  if (event && event->handler) {
    event->handler(event->data);
  }
  uv_close((uv_handle_t *)handle, NULL);
  free(event);
}

void async_callback(uv_async_t *handle) {
  dispatcher *disp = (dispatcher *)handle->data;

  timer_event *event;
  while ((event = (timer_event *)queue_pop(&disp->event_queue)) != NULL) {
    uv_timer_t *timer_handle = malloc(sizeof(uv_timer_t));
    uv_timer_init(disp->loop, timer_handle);
    timer_handle->data = event;

    uv_timer_start(timer_handle, timer_callback, event->timeout, 0);
  }
}

void dispatcher_init(dispatcher *disp) {
  disp->loop = malloc(sizeof(uv_loop_t));
  uv_loop_init(disp->loop);

  uv_async_init(disp->loop, &disp->async_req, async_callback);
  disp->async_req.data = disp;

  queue_init(&disp->event_queue);
}

void dispatcher_destroy(dispatcher *disp) {
  uv_close((uv_handle_t *)&disp->async_req, NULL);
  uv_loop_close(disp->loop);
  free(disp->loop);
}

void push_event(dispatcher *disp, event_handler handler, void *data,
                unsigned long timeout) {
  timer_event *event = malloc(sizeof(timer_event));
  event->handler = handler;
  event->data = data;
  event->timeout = timeout;

  queue_push(&disp->event_queue, event);
  uv_async_send(&disp->async_req);
}

void dispatcher_run(dispatcher *disp) {
  printf("Starting event loop...\n");
  uv_run(disp->loop, UV_RUN_DEFAULT);
}

void sample_handler(void *data) {
  time_t rawtime;
  struct tm *timeinfo;
  time(&rawtime);
  timeinfo = localtime(&rawtime);
  printf("Timer triggered! Data: %s, Time: %s", (char *)data,
         asctime(timeinfo));
}

void *dispatcher_thread_func(void *arg) {
  dispatcher *disp = (dispatcher *)arg;
  dispatcher_run(disp);
  return NULL;
}

int main(void) {
  dispatcher disp;
  dispatcher_init(&disp);

  pthread_t dispatcher_thread;
  pthread_create(&dispatcher_thread, NULL, dispatcher_thread_func, &disp);

  char data1[] = "Event 1";
  char data2[] = "Event 2";
  push_event(&disp, sample_handler, data1, 3000);

  sleep(1);
  
  push_event(&disp, sample_handler, data2, 1000);

  pthread_join(dispatcher_thread, NULL);

  dispatcher_destroy(&disp);
  return 0;
}

Makefile

CC = gcc
CFLAGS = -Wall -pedantic -std=gnu11 -O2
LDFLAGS = -luv -lpthread

SRC = dispatcher.c
OBJ = $(SRC:.c=.o)
EXEC = dispatcher

all: $(EXEC)
	
$(EXEC): $(OBJ)
	$(CC) -o $@ $^ $(LDFLAGS)

%.o: %.c
	$(CC) $(CFLAGS) -c -o $@ $<
	
clean:
	rm -f $(EXEC) $(OBJ)
	
.PHONY: clean