X-Git-Url: http://git.eversberg.eu/gitweb.cgi?p=lcr.git;a=blobdiff_plain;f=message.c;h=cfda36be28f839b268e4c79f67fb3119c0b490a4;hp=eab297803336c257823ebe19b74874db3a187f27;hb=refs%2Fheads%2Fbackup2;hpb=0a0643e3a561de5f34927ea80c2bd0ce6405fd3c diff --git a/message.c b/message.c index eab2978..cfda36b 100644 --- a/message.c +++ b/message.c @@ -9,22 +9,42 @@ ** ** \*****************************************************************************/ -#include -#include -#include #include "main.h" MESSAGES -struct message *message_first = NULL; -struct message **messagepointer_end = &message_first; +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 message *message_create(int id_from, int id_to, int flow, int type) +struct lcr_msg *message_create(int id_from, int id_to, int flow, int type) { - struct message *message; + 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 message *)MALLOC(sizeof(struct message)); + message = (struct lcr_msg *)MALLOC(sizeof(struct lcr_msg)); if (!message) FATAL("No memory for message.\n"); mmemuse++; @@ -38,28 +58,32 @@ struct message *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 message *message) +void _message_put(struct lcr_msg *message, const char *file, int line) { - if (message->id_to == 0) - { + if (message->id_to == 0) { PDEBUG(DEBUG_MSG, "message %s not written, because destination is 0.\n", messages_txt[message->type]); message_free(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 message *message_forward(int id_from, int id_to, int flow, union parameter *param) +struct lcr_msg *message_forward(int id_from, int id_to, int flow, union parameter *param) { - struct message *message; + struct lcr_msg *message; /* get point to message */ - message = (struct message *)((unsigned long)param - ((unsigned long)(&message->param) - (unsigned long)message)); + message = (struct lcr_msg *)((unsigned long)param - ((unsigned long)(&message->param) - (unsigned long)message)); /* protect, so forwarded messages are not freed after handling */ message->keep = 1; @@ -73,14 +97,12 @@ struct message *message_forward(int id_from, int id_to, int flow, union paramete } /* detaches the first messages from the message chain */ -struct message *message_get(void) +struct lcr_msg *message_get(void) { - struct message *message; + struct lcr_msg *message; if (!message_first) - { return(0); - } message = message_first; message_first = message->next; @@ -89,20 +111,82 @@ struct message *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); } /* free a message */ -void message_free(struct message *message) +void message_free(struct lcr_msg *message) { if (message->keep) return; - FREE(message, sizeof(struct message)); + FREE(message, sizeof(struct lcr_msg)); mmemuse--; } +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; +} +