]> code.citadel.org Git - citadel.git/blobdiff - citadel/serv_network.c
* You guessed it: still more code for the new networker.
[citadel.git] / citadel / serv_network.c
index 611b4a8b2e5c78ce938099e9ca7f1260025e769b..09e14c0d230da4fe8aa148fdc11510ad4faf015e 100644 (file)
@@ -18,7 +18,7 @@
 #include <pwd.h>
 #include <errno.h>
 #include <sys/types.h>
-
+#include <dirent.h>
 #if TIME_WITH_SYS_TIME
 # include <sys/time.h>
 # include <time.h>
@@ -64,6 +64,156 @@ struct RoomProcList {
 struct RoomProcList *rplist = NULL;
 
 
+/*
+ * We build a map of the Citadel network during network runs.
+ */
+struct NetMap {
+       struct NetMap *next;
+       char nodename[SIZ];
+       time_t lastcontact;
+       char nexthop[SIZ];
+};
+
+struct NetMap *the_netmap = NULL;
+
+
+
+/* 
+ * Read the network map from its configuration file into memory.
+ */
+void read_network_map(void) {
+       char *serialized_map = NULL;
+       int i;
+       char buf[SIZ];
+       struct NetMap *nmptr;
+
+       serialized_map = CtdlGetSysConfig(IGNETMAP);
+       if (serialized_map == NULL) return;     /* if null, no entries */
+
+       /* Use the string tokenizer to grab one line at a time */
+       for (i=0; i<num_tokens(serialized_map, '\n'); ++i) {
+               extract_token(buf, serialized_map, i, '\n');
+               nmptr = (struct NetMap *) mallok(sizeof(struct NetMap));
+               extract(nmptr->nodename, buf, 0);
+               nmptr->lastcontact = extract_long(buf, 1);
+               extract(nmptr->nexthop, buf, 2);
+               nmptr->next = the_netmap;
+               the_netmap = nmptr;
+       }
+
+       phree(serialized_map);
+}
+
+
+/*
+ * Write the network map from memory back to the configuration file.
+ */
+void write_network_map(void) {
+       char *serialized_map = NULL;
+       struct NetMap *nmptr;
+
+       serialized_map = strdoop("");
+
+       if (the_netmap != NULL) {
+               for (nmptr = the_netmap; nmptr != NULL; nmptr = nmptr->next) {
+                       serialized_map = reallok(serialized_map,
+                                               (strlen(serialized_map)+SIZ) );
+                       if (strlen(nmptr->nodename) > 0) {
+                               sprintf(&serialized_map[strlen(serialized_map)],
+                                       "%s|%ld|%s\n",
+                                       nmptr->nodename,
+                                       nmptr->lastcontact,
+                                       nmptr->nexthop);
+                       }
+               }
+       }
+
+       CtdlPutSysConfig(IGNETMAP, serialized_map);
+       phree(serialized_map);
+
+       /* Now free the list */
+       while (the_netmap != NULL) {
+               nmptr = the_netmap->next;
+               phree(the_netmap);
+               the_netmap = nmptr;
+       }
+}
+
+
+
+/* 
+ * Check the network map and determine whether the supplied node name is
+ * valid.  If it is not a neighbor node, supply the name of a neighbor node
+ * which is the next hop.
+ */
+int is_valid_node(char *nexthop, char *node) {
+       char *ignetcfg = NULL;
+       int i;
+       char linebuf[SIZ];
+       char buf[SIZ];
+       int retval;
+       struct NetMap *nmptr;
+
+       if (node == NULL) {
+               return(-1);
+       }
+
+       /*
+        * First try the neighbor nodes
+        */
+       ignetcfg = CtdlGetSysConfig(IGNETCFG);
+       if (ignetcfg == NULL) {
+               if (nexthop != NULL) {
+                       strcpy(nexthop, "");
+               }
+               return(-1);
+       }
+
+       retval = (-1);
+       if (nexthop != NULL) {
+               strcpy(nexthop, "");
+       }
+
+       /* Use the string tokenizer to grab one line at a time */
+       for (i=0; i<num_tokens(ignetcfg, '\n'); ++i) {
+               extract_token(linebuf, ignetcfg, i, '\n');
+               extract(buf, linebuf, 0);
+               if (!strcasecmp(buf, node)) {
+                       if (nexthop != NULL) {
+                               strcpy(nexthop, "");
+                       }
+                       retval = 0;
+               }
+       }
+
+       phree(ignetcfg);
+       if (retval == 0) {
+               return(retval);         /* yup, it's a direct neighbor */
+       }
+
+       /*      
+        * If we get to this point we have to see if we know the next hop
+        */
+       if (the_netmap != NULL) {
+               for (nmptr = the_netmap; nmptr != NULL; nmptr = nmptr->next) {
+                       if (!strcasecmp(nmptr->nodename, node)) {
+                               if (nexthop != NULL) {
+                                       strcpy(nexthop, nmptr->nexthop);
+                               }
+                               return(0);
+                       }
+               }
+       }
+
+       /*
+        * If we get to this point, the supplied node name is bogus.
+        */
+       lprintf(5, "Invalid node name <%s>\n", node);
+       return(-1);
+}
+
+
+
 
 
 void cmd_gnet(char *argbuf) {
@@ -132,6 +282,7 @@ void network_spool_msg(long msgnum, void *userdata) {
        struct SpoolControl *sc;
        struct namelist *nptr;
        int err;
+       int i;
        char *instr = NULL;
        char *newpath = NULL;
        size_t instr_len = SIZ;
@@ -140,6 +291,10 @@ void network_spool_msg(long msgnum, void *userdata) {
        struct ser_ret sermsg;
        FILE *fp;
        char filename[SIZ];
+       char buf[SIZ];
+       int bang = 0;
+       int send = 1;
+       int delete_after_send = 0;      /* Set to 1 to delete after spooling */
 
        sc = (struct SpoolControl *)userdata;
 
@@ -226,6 +381,17 @@ void network_spool_msg(long msgnum, void *userdata) {
                        }
                        msg->cm_fields['C'] = strdoop(CC->quickroom.QRname);
 
+                       /*
+                        * Determine if this message is set to be deleted
+                        * after sending out on the network
+                        */
+                       if (msg->cm_fields['S'] != NULL) {
+                               if (!strcasecmp(msg->cm_fields['S'],
+                                  "CANCEL")) {
+                                       delete_after_send = 1;
+                               }
+                       }
+
                        /* 
                         * Now serialize it for transmission
                         */
@@ -236,26 +402,49 @@ void network_spool_msg(long msgnum, void *userdata) {
                        for (nptr = sc->ignet_push_shares; nptr != NULL;
                            nptr = nptr->next) {
 
-                               /* FIXME check for valid node name */
-                               /* FIXME check for split horizon */
+                               send = 1;
 
-                               /* Send the message */
-                               sprintf(filename, "./network/spoolout/%s",
-                                       nptr->name);
-                               fp = fopen(filename, "ab");
-                               if (fp != NULL) {
-                                       fwrite(sermsg.ser, sermsg.len, 1, fp);
-                                       fclose(fp);
+                               /* Check for valid node name */
+                               if (is_valid_node(NULL, nptr->name) != 0) {
+                                       lprintf(3, "Invalid node <%s>\n",
+                                               nptr->name);
+                                       send = 0;
                                }
-                       }
 
+                               /* Check for split horizon */
+                               bang = num_tokens(msg->cm_fields['P'], '!');
+                               if (bang > 1) for (i=0; i<(bang-1); ++i) {
+                                       extract_token(buf, msg->cm_fields['P'],
+                                               i, '!');
+                                       if (!strcasecmp(buf, nptr->name)) {
+                                               send = 0;
+                                       }
+                               }
 
+                               /* Send the message */
+                               if (send == 1) {
+                                       sprintf(filename,
+                                               "./network/spoolout/%s",
+                                               nptr->name);
+                                       fp = fopen(filename, "ab");
+                                       if (fp != NULL) {
+                                               fwrite(sermsg.ser,
+                                                       sermsg.len, 1, fp);
+                                               fclose(fp);
+                                       }
+                               }
+                       }
                }
-
        }
 
        /* update lastsent */
        sc->lastsent = msgnum;
+
+       /* Delete this message if delete-after-send is set */
+       if (delete_after_send) {
+               CtdlDeleteMessages(CC->quickroom.QRname, msgnum, "");
+       }
+
 }
        
 
@@ -371,7 +560,210 @@ void network_queue_room(struct quickroom *qrbuf, void *data) {
        ptr->next = rplist;
        rplist = ptr;
 }
+
+
+/*
+ * Learn topology from path fields
+ */
+void network_learn_topology(char *node, char *path) {
+       char nexthop[SIZ];
+       struct NetMap *nmptr;
+
+       strcpy(nexthop, "");
+
+       if (num_tokens(path, '!') < 3) return;
+       for (nmptr = the_netmap; nmptr != NULL; nmptr = nmptr->next) {
+               if (!strcasecmp(nmptr->nodename, node)) {
+                       extract_token(nmptr->nexthop, path, 0, '!');
+                       nmptr->lastcontact = time(NULL);
+                       return;
+               }
+       }
+
+       /* If we got here then it's not in the map, so add it. */
+       nmptr = (struct NetMap *) mallok(sizeof (struct NetMap));
+       strcpy(nmptr->nodename, node);
+       nmptr->lastcontact = time(NULL);
+       extract_token(nmptr->nexthop, path, 0, '!');
+       nmptr->next = the_netmap;
+       the_netmap = nmptr;
+}
+
+
+
+/*
+ * Process a buffer containing a single message from a single file
+ * from the inbound queue 
+ */
+void network_process_buffer(char *buffer, long size) {
+       struct CtdlMessage *msg;
+       long pos;
+       int field;
+       int a;
+       int e = MES_LOCAL;
+       struct usersupp tempUS;
+       char recp[SIZ];
+       char target_room[ROOMNAMELEN];
+
+       /* Set default target room to trash */
+       strcpy(target_room, TWITROOM);
+
+       /* Load the message into memory */
+        msg = (struct CtdlMessage *) mallok(sizeof(struct CtdlMessage));
+        memset(msg, 0, sizeof(struct CtdlMessage));
+        msg->cm_magic = CTDLMESSAGE_MAGIC;
+        msg->cm_anon_type = buffer[1];
+        msg->cm_format_type = buffer[2];
+
+       for (pos = 3; pos < size; ++pos) {
+               field = buffer[pos];
+               msg->cm_fields[field] = strdoop(&buffer[pos+1]);
+               pos = pos + strlen(&buffer[(int)pos]);
+       }
+
+       /* Check for message routing */
+       if (msg->cm_fields['D'] != NULL) {
+               if (strcasecmp(msg->cm_fields['D'], config.c_nodename)) {
+
+                       /* FIXME route the message, stupid */
+
+               }
+       }
+
+       /* FIXME check to see if we already have this message */
+
+       /* Learn network topology from the path */
+       if ((msg->cm_fields['N'] != NULL) && (msg->cm_fields['P'] != NULL)) {
+               network_learn_topology(msg->cm_fields['N'], 
+                                       msg->cm_fields['P']);
+       }
+
+       /* Does it have a recipient?  If so, validate it... */
+       if (msg->cm_fields['R'] != NULL) {
+
+               safestrncpy(recp, msg->cm_fields['R'], sizeof(recp));
+
+                e = alias(recp);        /* alias and mail type */
+                if ((recp[0] == 0) || (e == MES_ERROR)) {
+
+                       /* FIXME bounce the msg */
+
+                }
+                else if (e == MES_LOCAL) {
+                        a = getuser(&tempUS, recp);
+                        if (a != 0) {
+                               /* FIXME bounce the msg */
+                        }
+                       else {
+                               MailboxName(target_room, &tempUS, MAILROOM);
+                       }
+                }
+        }
+
+       else if (msg->cm_fields['C'] != NULL) {
+               safestrncpy(target_room,
+                       msg->cm_fields['C'],
+                       sizeof target_room);
+       }
+
+       else if (msg->cm_fields['O'] != NULL) {
+               safestrncpy(target_room,
+                       msg->cm_fields['O'],
+                       sizeof target_room);
+       }
+
+       /* save the message into a room */
+       msg->cm_flags = CM_SKIP_HOOKS;
+        CtdlSaveMsg(msg, "", target_room, 0);
+       CtdlFreeMessage(msg);
+}
+
+
+/*
+ * Process a single message from a single file from the inbound queue 
+ */
+void network_process_message(FILE *fp, long msgstart, long msgend) {
+       long hold_pos;
+       long size;
+       char *buffer;
+
+       hold_pos = ftell(fp);
+       size = msgend - msgstart + 1;
+       buffer = mallok(size);
+       if (buffer != NULL) {
+               fseek(fp, msgstart, SEEK_SET);
+               fread(buffer, size, 1, fp);
+               network_process_buffer(buffer, size);
+               phree(buffer);
+       }
+
+       fseek(fp, hold_pos, SEEK_SET);
+}
+
+
+/*
+ * Process a single file from the inbound queue 
+ */
+void network_process_file(char *filename) {
+       FILE *fp;
+       long msgstart = (-1L);
+       long msgend = (-1L);
+       long msgcur = 0L;
+       int ch;
+
+       lprintf(7, "network: processing <%s>\n", filename);
+
+       fp = fopen(filename, "rb");
+       if (fp == NULL) {
+               lprintf(5, "Error opening %s: %s\n",
+                       filename, strerror(errno));
+               return;
+       }
+
+       /* Look for messages in the data stream and break them out */
+       while (ch = getc(fp), ch >= 0) {
        
+               if (ch == 255) {
+                       if (msgstart >= 0L) {
+                               msgend = msgcur - 1;
+                               network_process_message(fp, msgstart, msgend);
+                       }
+                       msgstart = msgcur;
+               }
+
+               ++msgcur;
+       }
+
+       msgend = msgcur - 1;
+       if (msgstart >= 0L) {
+               network_process_message(fp, msgstart, msgend);
+       }
+
+       fclose(fp);
+       unlink(filename);
+}
+
+
+/*
+ * Process anything in the inbound queue
+ */
+void network_do_spoolin(void) {
+       DIR *dp;
+       struct dirent *d;
+       char filename[SIZ];
+
+       dp = opendir("./network/spoolin");
+       if (dp == NULL) return;
+
+       while (d = readdir(dp), d != NULL) {
+               sprintf(filename, "./network/spoolin/%s", d->d_name);
+               network_process_file(filename);
+       }
+
+
+       closedir(dp);
+}
+
 
 /*
  * network_do_queue()
@@ -383,7 +775,6 @@ void network_do_queue(void) {
        static time_t last_run = 0L;
        struct RoomProcList *ptr;
 
-#define NETWORK_QUEUE_FREQUENCY 3600   /* one hour ... FIXME put in config */
        /*
         * Run no more frequently than once every n seconds
         */
@@ -399,6 +790,8 @@ void network_do_queue(void) {
        doing_queue = 1;
        last_run = time(NULL);
 
+       read_network_map();
+
        /* 
         * Go ahead and run the queue
         */
@@ -413,6 +806,11 @@ void network_do_queue(void) {
                phree(ptr);
        }
 
+       lprintf(7, "network: processing inbound queue\n");
+       network_do_spoolin();
+
+       write_network_map();
+
        lprintf(7, "network: queue run completed\n");
        doing_queue = 0;
 }