* This module handles shared rooms, inter-Citadel mail, and outbound
* mailing list processing.
*
- * Copyright (c) 2000-2011 by the citadel.org team
+ * Copyright (c) 2000-2012 by the citadel.org team
*
* This program is open source software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
#include "netmail.h"
#include "ctdl_module.h"
-/*
- * Learn topology from path fields
- */
-void network_learn_topology(char *node, char *path) {
- char nexthop[256];
- 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, '!', sizeof nmptr->nexthop);
- nmptr->lastcontact = time(NULL);
- ++netmap_changed;
- return;
- }
- }
-
- /* If we got here then it's not in the map, so add it. */
- nmptr = (NetMap *) malloc(sizeof (NetMap));
- strcpy(nmptr->nodename, node);
- nmptr->lastcontact = time(NULL);
- extract_token(nmptr->nexthop, path, 0, '!', sizeof nmptr->nexthop);
- nmptr->next = the_netmap;
- the_netmap = nmptr;
- ++netmap_changed;
-}
-
/*
* Batch up and send all outbound traffic from the current room
*/
-void network_spoolout_room(char *room_to_spool) {
+void network_spoolout_room(RoomProcList *room_to_spool,
+ HashList *working_ignetcfg,
+ HashList *the_netmap)
+{
char buf[SIZ];
char filename[PATH_MAX];
SpoolControl *sc;
* Normally this should never happen, but once in a while maybe a room gets
* queued for networking and then deleted before it can happen.
*/
- if (CtdlGetRoom(&CC->room, room_to_spool) != 0) {
- syslog(LOG_CRIT, "ERROR: cannot load <%s>\n", room_to_spool);
+ if (CtdlGetRoom(&CC->room, room_to_spool->name) != 0) {
+ syslog(LOG_CRIT, "ERROR: cannot load <%s>\n", room_to_spool->name);
return;
}
}
syslog(LOG_INFO, "Networking started for <%s>\n", CC->room.QRname);
+ sc->working_ignetcfg = working_ignetcfg;
+ sc->the_netmap = the_netmap;
+
/* If there are digest recipients, we have to build a digest */
if (sc->digestrecps != NULL) {
sc->digestfp = tmpfile();
* Process a buffer containing a single message from a single file
* from the inbound queue
*/
-void network_process_buffer(char *buffer, long size) {
+void network_process_buffer(char *buffer, long size, HashList *working_ignetcfg, HashList *the_netmap, int *netmap_changed)
+{
+ struct CitContext *CCC = CC;
+ StrBuf *Buf = NULL;
struct CtdlMessage *msg = NULL;
long pos;
int field;
char *oldpath = NULL;
char filename[PATH_MAX];
FILE *fp;
- char nexthop[SIZ];
+ const StrBuf *nexthop = NULL;
unsigned char firstbyte;
unsigned char lastbyte;
- syslog(LOG_DEBUG, "network_process_buffer() processing %ld bytes\n", size);
+ QN_syslog(LOG_DEBUG, "network_process_buffer() processing %ld bytes\n", size);
/* Validate just a little bit. First byte should be FF and * last byte should be 00. */
firstbyte = buffer[0];
lastbyte = buffer[size-1];
if ( (firstbyte != 255) || (lastbyte != 0) ) {
- syslog(LOG_ERR, "Corrupt message ignored. Length=%ld, firstbyte = %d, lastbyte = %d\n",
- size, firstbyte, lastbyte);
+ QN_syslog(LOG_ERR, "Corrupt message ignored. Length=%ld, firstbyte = %d, lastbyte = %d\n",
+ size, firstbyte, lastbyte);
return;
}
if (strcasecmp(msg->cm_fields['D'], config.c_nodename)) {
/* route the message */
- strcpy(nexthop, "");
- if (is_valid_node(nexthop, NULL, msg->cm_fields['D']) == 0) {
+ Buf = NewStrBufPlain(msg->cm_fields['D'], -1);
+ if (is_valid_node(&nexthop,
+ NULL,
+ Buf,
+ working_ignetcfg,
+ the_netmap) == 0)
+ {
/* prepend our node to the path */
if (msg->cm_fields['P'] != NULL) {
oldpath = msg->cm_fields['P'];
serialize_message(&sermsg, msg);
/* now send it */
- if (IsEmptyStr(nexthop)) {
- strcpy(nexthop, msg->cm_fields['D']);
+ if (StrLength(nexthop) == 0) {
+ nexthop = Buf;
}
- snprintf(filename,
- sizeof filename,
- "%s/%s@%lx%x",
- ctdl_netout_dir,
- nexthop,
- time(NULL),
- rand()
+ snprintf(filename,
+ sizeof filename,
+ "%s/%s@%lx%x",
+ ctdl_netout_dir,
+ ChrPtr(nexthop),
+ time(NULL),
+ rand()
);
- syslog(LOG_DEBUG, "Appending to %s\n", filename);
+ QN_syslog(LOG_DEBUG, "Appending to %s\n", filename);
fp = fopen(filename, "ab");
if (fp != NULL) {
fwrite(sermsg.ser, sermsg.len, 1, fp);
fclose(fp);
}
else {
- syslog(LOG_ERR, "%s: %s\n", filename, strerror(errno));
+ QN_syslog(LOG_ERR, "%s: %s\n", filename, strerror(errno));
}
free(sermsg.ser);
CtdlFreeMessage(msg);
+ FreeStrBuf(&Buf);
return;
}
else { /* invalid destination node name */
+ FreeStrBuf(&Buf);
network_bounce(msg,
"A message you sent could not be delivered due to an invalid destination node"
/* 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']);
+ network_learn_topology(msg->cm_fields['N'],
+ msg->cm_fields['P'],
+ the_netmap,
+ netmap_changed);
}
/* Is the sending node giving us a very persuasive suggestion about
"Please check the address and try sending the message again.\n");
msg = NULL;
free_recipients(recp);
- syslog(LOG_DEBUG, "Bouncing message due to invalid recipient address.\n");
+ QNM_syslog(LOG_DEBUG, "Bouncing message due to invalid recipient address.\n");
return;
}
strcpy(target_room, ""); /* no target room if mail */
/*
* Process a single message from a single file from the inbound queue
*/
-void network_process_message(FILE *fp, long msgstart, long msgend) {
+void network_process_message(FILE *fp,
+ long msgstart,
+ long msgend,
+ HashList *working_ignetcfg,
+ HashList *the_netmap,
+ int *netmap_changed)
+{
long hold_pos;
long size;
char *buffer;
if (buffer != NULL) {
fseek(fp, msgstart, SEEK_SET);
if (fread(buffer, size, 1, fp) > 0) {
- network_process_buffer(buffer, size);
+ network_process_buffer(buffer,
+ size,
+ working_ignetcfg,
+ the_netmap,
+ netmap_changed);
}
free(buffer);
}
/*
* Process a single file from the inbound queue
*/
-void network_process_file(char *filename) {
+void network_process_file(char *filename,
+ HashList *working_ignetcfg,
+ HashList *the_netmap,
+ int *netmap_changed)
+{
+ struct CitContext *CCC = CC;
FILE *fp;
long msgstart = (-1L);
long msgend = (-1L);
fp = fopen(filename, "rb");
if (fp == NULL) {
- syslog(LOG_CRIT, "Error opening %s: %s\n", filename, strerror(errno));
+ QN_syslog(LOG_CRIT, "Error opening %s: %s\n", filename, strerror(errno));
return;
}
fseek(fp, 0L, SEEK_END);
- syslog(LOG_INFO, "network: processing %ld bytes from %s\n", ftell(fp), filename);
+ QN_syslog(LOG_INFO, "network: processing %ld bytes from %s\n", ftell(fp), filename);
rewind(fp);
/* Look for messages in the data stream and break them out */
if (ch == 255) {
if (msgstart >= 0L) {
msgend = msgcur - 1;
- network_process_message(fp, msgstart, msgend);
+ network_process_message(fp,
+ msgstart,
+ msgend,
+ working_ignetcfg,
+ the_netmap,
+ netmap_changed);
}
msgstart = msgcur;
}
msgend = msgcur - 1;
if (msgstart >= 0L) {
- network_process_message(fp, msgstart, msgend);
+ network_process_message(fp,
+ msgstart,
+ msgend,
+ working_ignetcfg,
+ the_netmap,
+ netmap_changed);
}
fclose(fp);
/*
* Process anything in the inbound queue
*/
-void network_do_spoolin(void) {
+void network_do_spoolin(HashList *working_ignetcfg, HashList *the_netmap, int *netmap_changed)
+{
+ struct CitContext *CCC = CC;
DIR *dp;
struct dirent *d;
struct stat statbuf;
*/
if (stat(ctdl_netin_dir, &statbuf)) return;
if (statbuf.st_mtime == last_spoolin_mtime) {
- syslog(LOG_DEBUG, "network: nothing in inbound queue\n");
+ QNM_syslog(LOG_DEBUG, "network: nothing in inbound queue\n");
return;
}
last_spoolin_mtime = statbuf.st_mtime;
- syslog(LOG_DEBUG, "network: processing inbound queue\n");
+ QNM_syslog(LOG_DEBUG, "network: processing inbound queue\n");
/*
* Ok, there's something interesting in there, so scan it.
ctdl_netin_dir,
d->d_name
);
- network_process_file(filename);
+ network_process_file(filename,
+ working_ignetcfg,
+ the_netmap,
+ netmap_changed);
}
}
* Step 1: consolidate files in the outbound queue into one file per neighbor node
* Step 2: delete any files in the outbound queue that were for neighbors who no longer exist.
*/
-void network_consolidate_spoolout(void) {
+void network_consolidate_spoolout(HashList *working_ignetcfg, HashList *the_netmap)
+{
+ struct CitContext *CCC = CC;
+ IOBuffer IOB;
+ FDIOBuffer FDIO;
+ int d_namelen;
DIR *dp;
struct dirent *d;
+ struct dirent *filedir_entry;
+ const char *pch;
+ char spooloutfilename[PATH_MAX];
char filename[PATH_MAX];
- char cmd[PATH_MAX];
- char nexthop[256];
+ const StrBuf *nexthop;
+ StrBuf *NextHop;
int i;
- char *ptr;
+ struct stat statbuf;
+ int nFailed = 0;
/* Step 1: consolidate files in the outbound queue into one file per neighbor node */
+ d = (struct dirent *)malloc(offsetof(struct dirent, d_name) + PATH_MAX + 1);
+ if (d == NULL) return;
+
dp = opendir(ctdl_netout_dir);
- if (dp == NULL) return;
- while (d = readdir(dp), d != NULL) {
- if (
- (strcmp(d->d_name, "."))
- && (strcmp(d->d_name, ".."))
- && (strchr(d->d_name, '@') != NULL)
- ) {
- safestrncpy(nexthop, d->d_name, sizeof nexthop);
- ptr = strchr(nexthop, '@');
- if (ptr) *ptr = 0;
-
- snprintf(filename,
- sizeof filename,
- "%s/%s",
- ctdl_netout_dir,
- d->d_name
- );
-
- syslog(LOG_DEBUG, "Consolidate %s to %s\n", filename, nexthop);
- if (network_talking_to(nexthop, NTT_CHECK)) {
- syslog(LOG_DEBUG,
- "Currently online with %s - skipping for now\n",
- nexthop
+ if (dp == NULL) {
+ free(d);
+ return;
+ }
+
+ NextHop = NewStrBuf();
+ memset(&IOB, 0, sizeof(IOBuffer));
+ memset(&FDIO, 0, sizeof(FDIOBuffer));
+ FDIO.IOB = &IOB;
+
+ while ((readdir_r(dp, d, &filedir_entry) == 0) &&
+ (filedir_entry != NULL))
+ {
+#ifdef _DIRENT_HAVE_D_NAMELEN
+ d_namelen = filedir_entry->d_namelen;
+#else
+
+#ifndef DT_UNKNOWN
+#define DT_UNKNOWN 0
+#define DT_DIR 4
+#define DT_REG 8
+#define DT_LNK 10
+
+#define IFTODT(mode) (((mode) & 0170000) >> 12)
+#define DTTOIF(dirtype) ((dirtype) << 12)
+#endif
+ d_namelen = strlen(filedir_entry->d_name);
+#endif
+ if ((d_namelen > 1) && filedir_entry->d_name[d_namelen - 1] == '~')
+ continue; /* Ignore backup files... */
+
+ if ((d_namelen == 1) &&
+ (filedir_entry->d_name[0] == '.'))
+ continue;
+
+ if ((d_namelen == 2) &&
+ (filedir_entry->d_name[0] == '.') &&
+ (filedir_entry->d_name[1] == '.'))
+ continue;
+
+ pch = strchr(filedir_entry->d_name, '@');
+ if (pch == NULL)
+ continue;
+
+ snprintf(filename,
+ sizeof filename,
+ "%s/%s",
+ ctdl_netout_dir,
+ filedir_entry->d_name);
+
+ StrBufPlain(NextHop,
+ filedir_entry->d_name,
+ pch - filedir_entry->d_name);
+
+ snprintf(spooloutfilename,
+ sizeof spooloutfilename,
+ "%s/%s",
+ ctdl_netout_dir,
+ ChrPtr(NextHop));
+
+ QN_syslog(LOG_DEBUG, "Consolidate %s to %s\n", filename, ChrPtr(NextHop));
+ if (network_talking_to(SKEY(NextHop), NTT_CHECK)) {
+ nFailed++;
+ QN_syslog(LOG_DEBUG,
+ "Currently online with %s - skipping for now\n",
+ ChrPtr(NextHop)
);
+ }
+ else {
+ size_t dsize;
+ size_t fsize;
+ int infd, outfd;
+ const char *err = NULL;
+ network_talking_to(SKEY(NextHop), NTT_ADD);
+
+ infd = open(filename, O_RDONLY);
+ if (infd == -1) {
+ nFailed++;
+ QN_syslog(LOG_ERR,
+ "failed to open %s for reading due to %s; skipping.\n",
+ filename, strerror(errno)
+ );
+ network_talking_to(SKEY(NextHop), NTT_REMOVE);
+ continue;
+ }
+
+ outfd = open(spooloutfilename,
+ O_EXCL|O_CREAT|O_NONBLOCK|O_WRONLY,
+ S_IRUSR|S_IWUSR);
+ if (outfd == -1)
+ {
+ outfd = open(spooloutfilename,
+ O_EXCL|O_NONBLOCK|O_WRONLY,
+ S_IRUSR | S_IWUSR);
+ }
+ if (outfd == -1) {
+ nFailed++;
+ QN_syslog(LOG_ERR,
+ "failed to open %s for reading due to %s; skipping.\n",
+ spooloutfilename, strerror(errno)
+ );
+ close(infd);
+ network_talking_to(SKEY(NextHop), NTT_REMOVE);
+ continue;
+ }
+
+ dsize = lseek(outfd, 0, SEEK_END);
+ lseek(outfd, -dsize, SEEK_SET);
+
+ fstat(infd, &statbuf);
+ fsize = statbuf.st_size;
+/*
+ fsize = lseek(infd, 0, SEEK_END);
+*/
+ IOB.fd = infd;
+ FDIOBufferInit(&FDIO, &IOB, outfd, fsize + dsize);
+ FDIO.ChunkSendRemain = fsize;
+ FDIO.TotalSentAlready = dsize;
+ err = NULL;
+ errno = 0;
+ do {} while ((FileMoveChunked(&FDIO, &err) > 0) && (err == NULL));
+ if (err == NULL) {
+ unlink(filename);
}
else {
- network_talking_to(nexthop, NTT_ADD);
- snprintf(cmd, sizeof cmd, "/bin/cat %s >>%s/%s && /bin/rm -f %s",
- filename,
- ctdl_netout_dir, nexthop,
- filename
- );
- system(cmd);
- network_talking_to(nexthop, NTT_REMOVE);
+ nFailed++;
+ QN_syslog(LOG_ERR,
+ "failed to append to %s [%s]; rolling back..\n",
+ spooloutfilename, strerror(errno)
+ );
+ /* whoops partial append?? truncate spooloutfilename again! */
+ ftruncate(outfd, dsize);
}
+ FDIOBufferDelete(&FDIO);
+ close(infd);
+ close(outfd);
+ network_talking_to(SKEY(NextHop), NTT_REMOVE);
}
}
closedir(dp);
- /* Step 2: delete any files in the outbound queue that were for neighbors who no longer exist */
+ if (nFailed > 0) {
+ FreeStrBuf(&NextHop);
+ QN_syslog(LOG_INFO,
+ "skipping Spoolcleanup because of %d files unprocessed.\n",
+ nFailed
+ );
+
+ return;
+ }
+ /* Step 2: delete any files in the outbound queue that were for neighbors who no longer exist */
dp = opendir(ctdl_netout_dir);
- if (dp == NULL) return;
+ if (dp == NULL) {
+ FreeStrBuf(&NextHop);
+ free(d);
+ return;
+ }
- while (d = readdir(dp), d != NULL) {
- if (!strcmp(d->d_name, ".") || !strcmp(d->d_name, ".."))
+ while ((readdir_r(dp, d, &filedir_entry) == 0) &&
+ (filedir_entry != NULL))
+ {
+#ifdef _DIRENT_HAVE_D_NAMELEN
+ d_namelen = filedir_entry->d_namelen;
+ d_type = filedir_entry->d_type;
+#else
+
+#ifndef DT_UNKNOWN
+#define DT_UNKNOWN 0
+#define DT_DIR 4
+#define DT_REG 8
+#define DT_LNK 10
+
+#define IFTODT(mode) (((mode) & 0170000) >> 12)
+#define DTTOIF(dirtype) ((dirtype) << 12)
+#endif
+ d_namelen = strlen(filedir_entry->d_name);
+#endif
+ if ((d_namelen == 1) &&
+ (filedir_entry->d_name[0] == '.'))
continue;
- ptr = strchr(d->d_name, '@');
- if (d != NULL)
+
+ if ((d_namelen == 2) &&
+ (filedir_entry->d_name[0] == '.') &&
+ (filedir_entry->d_name[1] == '.'))
+ continue;
+
+ pch = strchr(filedir_entry->d_name, '@');
+ if (pch == NULL) /* no @ in name? consolidated file. */
continue;
+
+ StrBufPlain(NextHop,
+ filedir_entry->d_name,
+ pch - filedir_entry->d_name);
+
snprintf(filename,
sizeof filename,
"%s/%s",
ctdl_netout_dir,
- d->d_name
+ filedir_entry->d_name
);
- strcpy(nexthop, "");
- i = is_valid_node(nexthop, NULL, d->d_name);
+ i = is_valid_node(&nexthop,
+ NULL,
+ NextHop,
+ working_ignetcfg,
+ the_netmap);
- if ( (i != 0) || !IsEmptyStr(nexthop) ) {
+ if ( (i != 0) || (StrLength(nexthop) > 0) ) {
unlink(filename);
}
}
-
-
+ FreeStrBuf(&NextHop);
+ free(d);
closedir(dp);
}
if (!threading)
{
create_spool_dirs();
- CtdlRegisterCleanupHook(destroy_network_queue_room);
+//////todo CtdlRegisterCleanupHook(destroy_network_queue_room);
}
return "network_spool";
}
projects / citadel.git / blobdiff