ttt.c

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

#define THREAD_STATUS_BAR 0
#define THREAD_COMMUNICATION 1

#define THREAD_STATUS_DEAD 0
#define THREAD_STATUS_ALIVE 1
#define THREAD_STATUS_COMMAND 2
#define THREAD_STATUS_RESPONSE 3
#define THREAD_STATUS_PAUSE 4

#define THREAD_BLOCK 0
#define THREAD_FREE 1


typedef struct message {
  int len;
  char data[4096];
  struct message *front;
  struct message *rear;
} message;

typedef struct queueMessage {
  struct message *first;
  struct message *last;
  int sem;
  int size;
} queueMessage;

struct queueMessage *connThreadEvents[10];
/*struct queueMessage *connThreadEvents = malloc(sizeof(queueMessage) * 10);*/
/*queueMessage1 (*connThreadEvents)[];*/
/*typedef queueMessage list[10];*/
/*list *connThreadEvents = NULL;*/

void context_channel_sem_wait(queueMessage *threadControl)
{
  if (threadControl) {
    /*while(threadControl->sem == THREAD_BLOCK) sleep(1);*/
    while(threadControl->sem == THREAD_BLOCK);
    threadControl->sem = THREAD_BLOCK;
  }
}

void context_channel_sem_push(queueMessage *threadControl)
{
  if (threadControl) threadControl->sem = THREAD_FREE;
}

queueMessage *context_channel_new(void)
{
  queueMessage *queue = (queueMessage*) malloc(sizeof(queueMessage));

  queue->size = 0;
  queue->sem = THREAD_BLOCK;
  return queue;
}

int thread_channel_enqueue(struct queueMessage *queue, char *buf, int len)
{
  struct message *newMessage;

  if (len > 0) {
    context_channel_sem_wait(queue);
    newMessage = (message *)malloc(sizeof(message));

    /*Copy message*/
    memset(newMessage->data, 0, sizeof(newMessage->data));
    memcpy(newMessage->data, buf, len);
    newMessage->len = len;

    if (queue->size == 0) {
      queue->first = newMessage;
      queue->last = newMessage;
      queue->size = 1;
    } else {
      /*populate last, queue and node*/
      queue->last->rear = newMessage;
      newMessage->front = queue->last;
      queue->last = newMessage;
      queue->size++;
    }

    context_channel_sem_push(queue);
  }

  return len;
}

int thread_channel_dequeue(struct queueMessage *queue, char *buf)
{
  int len=0;
  struct message *first;
  struct message *local;

  if (queue->size > 0) {
    context_channel_sem_wait(queue);

    local = queue->first;

    memcpy(buf, local->data, local->len);

    len   = local->len;
    first = local->rear;

    queue->first = first;
    queue->size--;
    free(local);
    context_channel_sem_push(queue);
  }
  return len;
}

int subscribe(void) 
{
  /*int id = 0;*/
  /*static struct queueMessage *queue = NULL;*/

  /*while (connThreadEvents[id] != NULL) { id++; };*/
  printf("\nbefore new");

  queueMessage *queue = (queueMessage*) malloc(sizeof(queueMessage));
  printf("\nafter");

  queue->size = 0;
  queue->sem = THREAD_BLOCK;
  printf("\nafter2");

  connThreadEvents[0] = (void *)queue;
  printf("\nafter3");

  /*queue = context_channel_new();*/
  /*printf("\nbefore events id [%d]", queue);*/
  /*connThreadEvents[id] = queue;*/
  /*printf("context_channel_new id [%d]", id);*/
  return 0;
}

int publish(int id, char *buf, int len)
{
  if (connThreadEvents[id] != NULL)
    return thread_channel_enqueue(connThreadEvents[id], buf, len);
  else
    return 0;
}

int listen(int id, char *buf)
{
  if (connThreadEvents[id] != NULL)
    return thread_channel_dequeue(connThreadEvents[id], buf);
  else
    return 0;
}

int main(void)
{
  int id, ret;
  char buf[6]  = "event\0";
  char buf2[6] = {0};

  id = subscribe();
  printf("Subscribed id [%d]", id);
  printf("Publish id [%d] ret [%d]", id, publish(0, buf, strlen(buf)));
  ret = listen(0, buf2);
  printf("Listen id [%d] ret [%d] value [%s]", id, ret, buf2);
}