twali/src/upubsub.h

/* upubsub: a minimal publish/subscribe 1 header file library for pure ANSI C. */
#ifndef UPUBSUB_H_INCLUDED
#define UPUBSUB_H_INCLUDED

#ifndef UPUBSUB_LISTENER_AMOUNT
#define UPUBSUB_LISTENER_AMOUNT 64
#endif /* UPUBSUB_LISTENER_AMOUNT */

#ifdef UPUBSUB_TEST

struct upubsub_test_message_type {
    char text[64];
    int x;
    int y;
};

#define UPUBSUB_MESSAGE_TYPE struct upubsub_test_message_type

#endif

#ifndef UPUBSUB_MESSAGE_SIZE
#ifndef UPUBSUB_MESSAGE_TYPE
#define UPUBSUB_MESSAGE_SIZE 64
#else 
#define UPUBSUB_MESSAGE_SIZE (sizeof(UPUBSUB_MESSAGE_TYPE))
#endif
#endif /* UPUBSUB_MESSAGE_SIZE */

#ifndef UPUBSUB_TOPIC_AMOUNT
#define UPUBSUB_TOPIC_AMOUNT 32
#endif /* UPUBSUB_TOPIC_AMOUNT */

#ifdef __cplusplus
extern "C" {
#endif

#include <stdlib.h>
#include <stdint.h>
#include <string.h>


enum upubsub_message_data_type {
    upubsub_message_data_type_bytes = 1,
    upubsub_message_data_type_ptr   = 2,
    upubsub_message_data_type_typed = 3,
};

struct upubsub_message {
	const char * topic;
	size_t	     size;
    enum upubsub_message_data_type data_type;
    union {
        uint8_t   bytes[UPUBSUB_MESSAGE_SIZE];
        void    * ptr;
#ifdef UPUBSUB_MESSAGE_TYPE
        UPUBSUB_MESSAGE_TYPE typed;
#endif
    } data;
};

struct upubsub_listener;
struct upubsub;

typedef int (upubsub_listen_func)(struct upubsub_listener *l, struct upubsub_message m);

struct upubsub_listener {
	struct upubsub * upubsub;
	unsigned int id;
	const char * topic;
	void * data;
	upubsub_listen_func * listen;
};

struct upubsub {
	size_t size;
	unsigned int last_id;
	struct upubsub_listener listeners[UPUBSUB_LISTENER_AMOUNT];
	size_t topics_size;
};


int upubsub_listener_cmp_topic_id(const void * v1, const void * v2);
int upubsub_listener_cmp_topic(const void * v1, const void * v2);
int upubsub_listener_cmp_id(const void * v1, const void * v2);
struct upubsub_listener * upubsub_subscribe_listener(struct upubsub * u, struct upubsub_listener l);
struct upubsub_listener* upubsub_subscribe(struct upubsub * u, const char * topic, void * data, upubsub_listen_func * listen);
int upubsub_publish_message(struct upubsub u, struct upubsub_message m);
int upubsub_publish_data(struct upubsub u, const char * topic, void * data, size_t size);
int upubsub_publish_str(struct upubsub u, const char * topic, char * str);
int upubsub_unsubscribe_listener(struct upubsub * u, struct upubsub_listener * l);

#ifdef UPUBSUB_MESSAGE_TYPE
int upubsub_publish_type(struct upubsub u, const char * topic, UPUBSUB_MESSAGE_TYPE m);
#endif

#ifdef __cplusplus
}
#endif


#endif /* UPUBSUB_H_INCLUDED */

#ifdef UPUBSUB_TEST
#define UPUBSUB_IMPLEMENTATION
#endif

#ifdef UPUBSUB_IMPLEMENTATION

#ifdef __cplusplus
#error This is not C++. Please compile this with a C compiler.
#endif

/** Result of searching with brange_search. If high < low, nothing was found. */
struct bsearch_range_result {
    int low;
    int high;
};

struct bsearch_range_result bsearch_range (const void *key, const void *base, size_t nel, size_t width, int (*cmp)(const void *, const void *)) {
    struct bsearch_range_result res;
	void *here;
	int sign;
    int botm = 0;
    int top = nel;    
    res.low = 0;
    res.high = nel;
	while (res.low <= top) {
        int mid = res.low + (top - res.low) / 2;
        here = (char*)base + (mid*width);
		sign = cmp(key, here);
		if (sign <= 0) {
			top = mid - 1;
		} else {
			res.low = mid + 1;
		}
	}
    botm = res.low ;
	while (botm <= res.high) {
        int mid = botm + (res.high - botm) / 2;
        here = (char*)base + (mid*width);
		sign = cmp(key, here);
		if (sign < 0) {
			res.high = mid - 1;
		} else {
			botm = mid + 1;
		}
	}
	return res;
}



int upubsub_listener_cmp_topic_id(const void * v1, const void * v2) {
	int res;
	struct upubsub_listener *l1, *l2;
	l1 = (struct upubsub_listener *)(v1);
	l2 = (struct upubsub_listener *)(v2);

	/* Sort with null topic as last, as these ones are unused. */
	if ((!l1->topic) && (l2->topic)) { return 1;  }
	if ((!l2->topic) && (l1->topic)) { return -1; }
	if ((!l2->topic) && (!l1->topic)) { return 0; }
     	/* Otherwise compare topics. */
	res = strcmp(l1->topic, l2->topic);
	/* If the same sort by id to try to keep the sort stable. */
	 if (!res) {
		return l1->id - l2->id;
	}
	return res;
}

int upubsub_listener_cmp_topic(const void * v1, const void * v2) {
       struct upubsub_listener *l1, *l2;
       l1 = (struct upubsub_listener *)(v1);
       l2 = (struct upubsub_listener *)(v2);

       /* Sort with null topic as last, as these ones are unused. */
       if ((!l1->topic) && (l2->topic)) { return 1;  }
       if ((!l2->topic) && (l1->topic)) { return -1; }
       if ((!l2->topic) && (!l1->topic)) { return 0; }
       /* Otherwise compare topics. */
       return strcmp(l1->topic, l2->topic);
}

int upubsub_listener_cmp_id(const void * v1, const void * v2) {
	struct upubsub_listener *l1, *l2;
	l1 = (struct upubsub_listener *)(v1);
	l2 = (struct upubsub_listener *)(v2);

	/* Sort with null topic as last, as these ones are unused. */
	if ((!l1->topic) && (l2->topic)) { return 1;  }
	if ((!l2->topic) && (l1->topic)) { return -1; }
	if ((!l2->topic) && (!l1->topic)) { return 0; }
	/* Compare by id to find the exact listener. */
	return l1->id - l2->id;
}

struct upubsub_listener * upubsub_subscribe_listener(struct upubsub * u, struct upubsub_listener l)  {
	struct upubsub_listener * res;
	if ( u->size >= UPUBSUB_LISTENER_AMOUNT ) {
		return NULL;
	}
	u->last_id++;
	l.id = u->last_id;
	l.upubsub = u;
	u->listeners[u->size] = l;
	res = u->listeners + u->size;
	u->size ++;
	qsort(u->listeners, u->size, sizeof(struct upubsub_listener), upubsub_listener_cmp_topic_id);
	return res;
}

struct upubsub_listener* upubsub_subscribe(struct upubsub * u, const char * topic, void * data, upubsub_listen_func * listen) {
	struct upubsub_listener l;
	l.topic = topic;
	l.data = data;
	l.listen = listen;
	return upubsub_subscribe_listener(u, l);
}


int upubsub_publish_message(struct upubsub u, struct upubsub_message m) {
	int sent = 0;
	size_t index;
	struct upubsub_listener * found = NULL;
	struct upubsub_listener key;
    struct bsearch_range_result ran;

	key.topic = m.topic;
	key.id = 1;    
	ran = bsearch_range(&key, u.listeners, u.size, sizeof(struct upubsub_listener), upubsub_listener_cmp_topic);    
	if (ran.high < ran.low) { /* Not found, bail out */
		return sent;
	}
    /* Broadcast to all in range. */
	for (index = ran.low; index <= ran.high; index ++) {
		found = u.listeners + index;
		found->listen(found, m);
		sent ++;
	}
	return sent;
}

struct upubsub_message upubsub_message_make(const char * topic, void * data, size_t size) {
   	struct upubsub_message m = { 0 };
	m.size = size;
    if (m.size <= UPUBSUB_MESSAGE_SIZE) {
		memcpy(m.data.bytes, data, m.size);
		m.data_type = upubsub_message_data_type_bytes;
	} else {
        m.data.ptr = data;
        m.data_type = upubsub_message_data_type_ptr;
    }
	m.topic = topic;
    return m;
}

int upubsub_publish_data(struct upubsub u, const char * topic, void * data, size_t size) {
	struct upubsub_message m = upubsub_message_make(topic, data, size);
	return upubsub_publish_message(u, m);
}

int upubsub_publish_str(struct upubsub u, const char * topic, char * str) {
	return upubsub_publish_data(u, topic, str, strlen(str));
}


#ifdef UPUBSUB_MESSAGE_TYPE
int upubsub_publish_type(struct upubsub u, const char * topic, UPUBSUB_MESSAGE_TYPE mt) {
    struct upubsub_message m = { 0 };
	m.size = sizeof(mt);
    m.data.typed = mt;
    m.data_type = upubsub_message_data_type_typed;
	m.topic = topic;
	return upubsub_publish_message(u, m);
}
#endif


int upubsub_unsubscribe_listener(struct upubsub * u, struct upubsub_listener * l)  {
	struct upubsub_listener * found = NULL;

	found = bsearch(l, u->listeners, u->size, sizeof(struct upubsub_listener), upubsub_listener_cmp_id);
	if (!found) {
		return -1;
	}
	/* Set to unused. */
	found->id = 0;
	found->topic = NULL;
        /* Unused will be sorted to the back. */
	qsort(u->listeners, u->size, sizeof(struct upubsub_listener), upubsub_listener_cmp_topic_id);
	u->size--;
	return 0;
}

int upubsub_unsubscribe(struct upubsub_listener * l)  {
	return upubsub_unsubscribe_listener(l->upubsub, l);
}


#endif /* UPUBSUB_IMPLEMENTATION */


#ifdef UPUBSUB_TEST

#include <stdio.h>

int listen_print_str(struct upubsub_listener *l, struct upubsub_message m) {
	printf("received message on topic %s: data %s, %s\n", l->topic,
        (char*)(l->data), (char *)(m.data.bytes));
	return 0;
}

int listen_print_typed_str(struct upubsub_listener *l, struct upubsub_message m) {
	printf("received typed message on topic %s: data %s, %s, %d, %d\n", l->topic, 
        (char*)(l->data), (char *)(m.data.typed.text), 
        m.data.typed.x, m.data.typed.y);
	return 0;
}

int compare_int(const void * key, const void * mem) {
    int * i1 = (int*)key;
    int * i2 = (int*)mem;
    return (*i1) - (*i2);
}

int main(void) {
	struct upubsub pusu = { 0 };
	struct upubsub_listener *l[10];
    int arr[] = { 7, 8 , 8, 9, 9, 9, 10, 11, 12, 20, 20, 32 };
    int low, high;
    int key = 9;
    struct bsearch_range_result ran;
    ran = bsearch_range(&key, arr, 12, sizeof(int), compare_int);
    printf("Low, high: %d -> %d...\n ", ran.low, ran.high);
    if ((ran.low >= 0) && (ran.low < 12)) {
        printf("Low: %d\n", arr[ran.low]);
    }
    if ((ran.high >= 0) && (ran.high < 12)) {
        printf("High: %d\n", arr[ran.high]);
    }
    
    for (key = 0; key < 40; key ++) {     
        ran = bsearch_range(&key, arr, 12, sizeof(int), compare_int);
        printf("Key: %d; Low, high: %d -> %d:", key, ran.low, ran.high);
        if ((ran.low >= 0) && (ran.low < 12)) {
            printf(" Low: %d", arr[low]);
        }
        if ((ran.high >= 0) && (ran.high < 12)) {
            printf(" High: %d", arr[ran.high]);
        }
        printf("\n");
    }
    
    

	printf("Subscribing...\n");
	l[0] = upubsub_subscribe(&pusu, "POWER" , "SYST", listen_print_str);
	l[1] = upubsub_subscribe(&pusu, "POWER" , "APP1", listen_print_str);
	l[2] = upubsub_subscribe(&pusu, "SYSTEM", "APP2", listen_print_str);
	l[3] = upubsub_subscribe(&pusu, "SYSTEM", "WIND", listen_print_str);
    l[4] = upubsub_subscribe(&pusu, "SYSTEM", "APP3", listen_print_str);
	l[5] = upubsub_subscribe(&pusu, "POWER" , "APP4", listen_print_str);
    l[6] = upubsub_subscribe(&pusu, "TYPED" , "APP5", listen_print_typed_str);


	printf("Publishing...\n");
	upubsub_publish_str(pusu, "POWER", "DOWN");
	upubsub_publish_str(pusu, "POWER", "UP");
	upubsub_publish_str(pusu, "SYSTEM", "CLICK(7,8)");
	upubsub_publish_str(pusu, "SYSTEM", "ROLL(9,5)");
	upubsub_publish_str(pusu, "NULL", "IGNORE");
    {
        UPUBSUB_MESSAGE_TYPE typed = {0};
        strcpy(typed.text, "TYPED");
        typed.x = 10;
        typed.y = 20;
        upubsub_publish_type(pusu, "TYPED", typed);
    }

	printf("Removing topics...\n");
	upubsub_unsubscribe(l[1]);
	upubsub_unsubscribe(l[3]);

	printf("Publishing...\n");
	upubsub_publish_str(pusu, "POWER", "DOWN 2");
	upubsub_publish_str(pusu, "POWER", "UP 2");
	upubsub_publish_str(pusu, "SYSTEM", "CLICK(7,8) 2");
	upubsub_publish_str(pusu, "SYSTEM", "ROLL(9,5) 2");
	upubsub_publish_str(pusu, "NULL", "IGNORE");
    {
        UPUBSUB_MESSAGE_TYPE typed = {0};
        strcpy(typed.text, "TYPED");
        typed.x = 30;
        typed.y = 40;
        upubsub_publish_type(pusu, "TYPED", typed);
    }

	return 0;
}

#endif /* UPUBSUB_TEST */