fixup

[lcr.git] / message.c

diff --git a/message.c b/message.c
index 403253d..cfda36b 100644 (file)
--- a/message.c
+++ b/message.c
@@ -15,11 +15,34 @@ MESSAGES
 
 struct lcr_msg *message_first = NULL;
 struct lcr_msg **messagepointer_end = &message_first;
+struct lcr_work message_work;
+
+static int work_message(struct lcr_work *work, void *instance, int index);
+
+void init_message(void)
+{
+       memset(&message_work, 0, sizeof(message_work));
+       add_work(&message_work, work_message, NULL, 0);
+}
+
+void cleanup_message(void)
+{
+       del_work(&message_work);
+}
+
+unsigned int lcr_random = 0;
 
 /* creates a new message with the given attributes. the message must be filled then. after filling, the message_put must be called */
 struct lcr_msg *message_create(int id_from, int id_to, int flow, int type)
 {
        struct lcr_msg *message;
+       struct timeval now_tv;
+       struct timezone now_tz;
+
+       gettimeofday(&now_tv, &now_tz);
+       lcr_random = (lcr_random << 1) | (lcr_random >> 31);
+       lcr_random ^= now_tv.tv_sec;
+       lcr_random ^= now_tv.tv_usec;
 
        message = (struct lcr_msg *)MALLOC(sizeof(struct lcr_msg));
        if (!message)
@@ -35,7 +58,7 @@ struct lcr_msg *message_create(int id_from, int id_to, int flow, int type)
 }
 
 /* attaches a message to the end of the message chain */
-void message_put(struct lcr_msg *message)
+void _message_put(struct lcr_msg *message, const char *file, int line)
 {
        if (message->id_to == 0) {
                PDEBUG(DEBUG_MSG, "message %s not written, because destination is 0.\n", messages_txt[message->type]);
@@ -43,13 +66,16 @@ void message_put(struct lcr_msg *message)
                return;
        }
        
-       if ((options.deb&DEBUG_MSG) && message->type != MESSAGE_DATA)
-               PDEBUG(DEBUG_MSG, "message %s written from %ld to %ld (memory %x)\n", messages_txt[message->type], message->id_from, message->id_to, message);
+       if ((options.deb & DEBUG_MSG))
+               PDEBUG(DEBUG_MSG, "message %s written from %ld to %ld (memory %x at file %s, line %d)\n", messages_txt[message->type], message->id_from, message->id_to, message, file, line);
 
        *messagepointer_end = message;
        messagepointer_end = &(message->next);
        /* Nullify next pointer if recycled messages */
        *messagepointer_end=NULL;
+
+       /* trigger work */
+       trigger_work(&message_work);
 }
 
 struct lcr_msg *message_forward(int id_from, int id_to, int flow, union parameter *param)
@@ -85,8 +111,7 @@ struct lcr_msg *message_get(void)
 
        message->keep = 0;
 
-       if ((options.deb&DEBUG_MSG) && message->type != MESSAGE_DATA)
-
+       if ((options.deb & DEBUG_MSG))
                PDEBUG(DEBUG_MSG, "message %s reading from %ld to %ld (memory %x)\n", messages_txt[message->type], message->id_from, message->id_to, message);
 
        return(message);
@@ -102,3 +127,66 @@ void message_free(struct lcr_msg *message)
 }
 
 
+static int work_message(struct lcr_work *work, void *instance, int index)
+{
+       struct lcr_msg          *message;
+       class Port              *port;
+       class Endpoint          *epoint;
+       class Join              *join;
+
+       while ((message = message_get())) {
+               switch(message->flow) {
+                       case PORT_TO_EPOINT:
+                       epoint = find_epoint_id(message->id_to);
+                       if (epoint) {
+                               if (epoint->ep_app) {
+                                       epoint->ep_app->ea_message_port(message->id_from, message->type, &message->param);
+                               } else {
+                                       PDEBUG(DEBUG_MSG, "Warning: message %s from port %d to endpoint %d. endpoint doesn't have an application.\n", messages_txt[message->type], message->id_from, message->id_to);
+                               }
+                       } else {
+                               PDEBUG(DEBUG_MSG, "Warning: message %s from port %d to endpoint %d. endpoint doesn't exist anymore.\n", messages_txt[message->type], message->id_from, message->id_to);
+                       }
+                       break;
+
+                       case EPOINT_TO_JOIN:
+                       join = find_join_id(message->id_to);
+                       if (join) {
+                               join->message_epoint(message->id_from, message->type, &message->param);
+                       } else {
+                               PDEBUG(DEBUG_MSG, "Warning: message %s from endpoint %d to join %d. join doesn't exist anymore\n", messages_txt[message->type], message->id_from, message->id_to);
+                       }
+                       break;
+
+                       case JOIN_TO_EPOINT:
+                       epoint = find_epoint_id(message->id_to);
+                       if (epoint) {
+                               if (epoint->ep_app) {
+                                       epoint->ep_app->ea_message_join(message->id_from, message->type, &message->param);
+                               } else {
+                                       PDEBUG(DEBUG_MSG, "Warning: message %s from join %d to endpoint %d. endpoint doesn't have an application.\n", messages_txt[message->type], message->id_from, message->id_to);
+                               }
+                       } else {
+                               PDEBUG(DEBUG_MSG, "Warning: message %s from join %d to endpoint %d. endpoint doesn't exist anymore.\n", messages_txt[message->type], message->id_from, message->id_to);
+                       }
+                       break;
+
+                       case EPOINT_TO_PORT:
+                       port = find_port_id(message->id_to);
+                       if (port) {
+                               port->message_epoint(message->id_from, message->type, &message->param);
+BUDETECT
+                       } else {
+                               PDEBUG(DEBUG_MSG, "Warning: message %s from endpoint %d to port %d. port doesn't exist anymore\n", messages_txt[message->type], message->id_from, message->id_to);
+                       }
+                       break;
+
+                       default:
+                       PERROR("Message flow %d unknown.\n", message->flow);
+               }
+               message_free(message);
+       }
+
+       return 0;
+}
+