X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Fnetwork%2Fserv_networkclient.c;h=996d4e042bc72ad92ca1772bbdec2f49bc541638;hb=a0eac8aaf455e61bdc60bca946ceabb9e549c85d;hp=455e47f02e8e2bda9903b0f810143ddcfba55942;hpb=c06372f95e9313f29a331c7d76ef812e480f3c84;p=citadel.git diff --git a/citadel/modules/network/serv_networkclient.c b/citadel/modules/network/serv_networkclient.c index 455e47f02..996d4e042 100644 --- a/citadel/modules/network/serv_networkclient.c +++ b/citadel/modules/network/serv_networkclient.c @@ -2,7 +2,7 @@ * 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 @@ -88,6 +88,32 @@ struct CitContext networker_client_CC; +#define NODE ChrPtr(((AsyncNetworker*)IO->Data)->node) +#define N ((AsyncNetworker*)IO->Data)->n + +int NetworkClientDebugEnabled = 0; + +#define NCDBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (NetworkClientDebugEnabled != 0)) + +#define EVN_syslog(LEVEL, FORMAT, ...) \ + NCDBGLOG(LEVEL) syslog(LEVEL, \ + "IO[%ld]CC[%d]NW[%s][%ld]" FORMAT, \ + IO->ID, CCID, NODE, N, __VA_ARGS__) + +#define EVNM_syslog(LEVEL, FORMAT) \ + NCDBGLOG(LEVEL) syslog(LEVEL, \ + "IO[%ld]CC[%d]NW[%s][%ld]" FORMAT, \ + IO->ID, CCID, NODE, N) + +#define EVNCS_syslog(LEVEL, FORMAT, ...) \ + NCDBGLOG(LEVEL) syslog(LEVEL, "IO[%ld]NW[%s][%ld]" FORMAT, \ + IO->ID, NODE, N, __VA_ARGS__) + +#define EVNCSM_syslog(LEVEL, FORMAT) \ + NCDBGLOG(LEVEL) syslog(LEVEL, "IO[%ld]NW[%s][%ld]" FORMAT, \ + IO->ID, NODE, N) + + typedef enum _eNWCState { eeGreating, eAuth, @@ -125,36 +151,83 @@ eNextState nwc_connect_ip(AsyncIO *IO); eNextState NWC_SendQUIT(AsyncNetworker *NW); eNextState NWC_DispatchWriteDone(AsyncIO *IO); - -void DestroyNetworker(AsyncNetworker *NW) +void DeleteNetworker(void *vptr) { + AsyncNetworker *NW = (AsyncNetworker *)vptr; + FreeStrBuf(&NW->SpoolFileName); + FreeStrBuf(&NW->tempFileName); + FreeStrBuf(&NW->node); + FreeStrBuf(&NW->host); + FreeStrBuf(&NW->port); + FreeStrBuf(&NW->secret); + FreeStrBuf(&NW->Url); + FreeStrBuf(&NW->IO.ErrMsg); + FreeAsyncIOContents(&NW->IO); + if (NW->HostLookup.VParsedDNSReply != NULL) { + NW->HostLookup.DNSReplyFree(NW->HostLookup.VParsedDNSReply); + NW->HostLookup.VParsedDNSReply = NULL; + } + free(NW); } -#define NWC_DBG_SEND() syslog(LOG_DEBUG, "NW client[%ld]: > %s", NW->n, ChrPtr(NW->IO.SendBuf.Buf)) -#define NWC_DBG_READ() syslog(LOG_DEBUG, "NW client[%ld]: < %s\n", NW->n, ChrPtr(NW->IO.IOBuf)) +#define NWC_DBG_SEND() EVN_syslog(LOG_DEBUG, ": > %s", ChrPtr(NW->IO.SendBuf.Buf)) +#define NWC_DBG_READ() EVN_syslog(LOG_DEBUG, ": < %s\n", ChrPtr(NW->IO.IOBuf)) #define NWC_OK (strncasecmp(ChrPtr(NW->IO.IOBuf), "+OK", 3) == 0) +eNextState SendFailureMessage(AsyncIO *IO) +{ + AsyncNetworker *NW = IO->Data; + long lens[2]; + const char *strs[2]; + + strs[0] = ChrPtr(NW->node); + lens[0] = StrLength(NW->node); + + strs[1] = ChrPtr(NW->IO.ErrMsg); + lens[1] = StrLength(NW->IO.ErrMsg); + CtdlAideFPMessage( + ChrPtr(NW->IO.ErrMsg), + "Networker error", + 2, strs, (long*) &lens); + + return eAbort; +} + +eNextState FinalizeNetworker(AsyncIO *IO) +{ + AsyncNetworker *NW = (AsyncNetworker *)IO->Data; + + network_talking_to(SKEY(NW->node), NTT_REMOVE); + + DeleteNetworker(IO->Data); + return eAbort; +} + eNextState NWC_ReadGreeting(AsyncNetworker *NW) { char connected_to[SIZ]; + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); /* Read the server greeting */ /* Check that the remote is who we think it is and warn the Aide if not */ extract_token (connected_to, ChrPtr(NW->IO.IOBuf), 1, ' ', sizeof connected_to); if (strcmp(connected_to, ChrPtr(NW->node)) != 0) { + if (NW->IO.ErrMsg == NULL) + NW->IO.ErrMsg = NewStrBuf(); StrBufPrintf(NW->IO.ErrMsg, "Connected to node \"%s\" but I was expecting to connect to node \"%s\".", connected_to, ChrPtr(NW->node)); - syslog(LOG_ERR, "%s\n", ChrPtr(NW->IO.ErrMsg)); - CtdlAideMessage(ChrPtr(NW->IO.ErrMsg), "Network error"); - return eAbort;/// todo: aide message in anderer queue speichern + EVN_syslog(LOG_ERR, "%s\n", ChrPtr(NW->IO.ErrMsg)); + StopClientWatchers(IO, 1); + return QueueDBOperation(IO, SendFailureMessage); } return eSendReply; } eNextState NWC_SendAuth(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; /* We're talking to the correct node. Now identify ourselves. */ StrBufPrintf(NW->IO.SendBuf.Buf, "NETP %s|%s\n", config.c_nodename, @@ -165,6 +238,7 @@ eNextState NWC_SendAuth(AsyncNetworker *NW) eNextState NWC_ReadAuthReply(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] == '2') { @@ -172,33 +246,46 @@ eNextState NWC_ReadAuthReply(AsyncNetworker *NW) } else { + int Error = atol(ChrPtr(NW->IO.IOBuf)); + if (NW->IO.ErrMsg == NULL) + NW->IO.ErrMsg = NewStrBuf(); StrBufPrintf(NW->IO.ErrMsg, - "Connected to node \"%s\" but my secret wasn't accurate.", - ChrPtr(NW->node)); - syslog(LOG_ERR, "%s\n", ChrPtr(NW->IO.ErrMsg)); - CtdlAideMessage(ChrPtr(NW->IO.ErrMsg), "Network error"); - + "Connected to node \"%s\" but my secret wasn't accurate.\nReason was:%s\n", + ChrPtr(NW->node), ChrPtr(NW->IO.IOBuf) + 4); + if (Error == 552) { + EVN_syslog(LOG_INFO, + "Already talking to %s; skipping this time.\n", + ChrPtr(NW->node)); + + } + else { + EVN_syslog(LOG_ERR, "%s\n", ChrPtr(NW->IO.ErrMsg)); + StopClientWatchers(IO, 1); + return QueueDBOperation(IO, SendFailureMessage); + } return eAbort; } } eNextState NWC_SendNDOP(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NW->tempFileName = NewStrBuf(); NW->SpoolFileName = NewStrBuf(); - StrBufPrintf(NW->tempFileName, + StrBufPrintf(NW->SpoolFileName, "%s/%s.%lx%x", ctdl_netin_dir, ChrPtr(NW->node), time(NULL),// TODO: get time from libev rand()); - StrBufPrintf(NW->SpoolFileName, + StrBufStripSlashes(NW->SpoolFileName, 1); + StrBufPrintf(NW->tempFileName, "%s/%s.%lx%x", ctdl_nettmp_dir, ChrPtr(NW->node), time(NULL),// TODO: get time from libev rand()); - + StrBufStripSlashes(NW->tempFileName, 1); /* We're talking to the correct node. Now identify ourselves. */ StrBufPlain(NW->IO.SendBuf.Buf, HKEY("NDOP\n")); NWC_DBG_SEND(); @@ -207,6 +294,7 @@ eNextState NWC_SendNDOP(AsyncNetworker *NW) eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; int TotalSendSize; NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] == '2') @@ -214,20 +302,20 @@ eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) NW->IO.IOB.TotalSentAlready = 0; TotalSendSize = atol (ChrPtr(NW->IO.IOBuf) + 4); - syslog(LOG_DEBUG, "Expecting to transfer %ld bytes\n", NW->IO.IOB.TotalSendSize); + EVN_syslog(LOG_DEBUG, "Expecting to transfer %d bytes\n", TotalSendSize); if (TotalSendSize <= 0) { NW->State = eNUOP - 1; } else { int fd; - fd = open(ChrPtr(NW->SpoolFileName), + fd = open(ChrPtr(NW->tempFileName), O_EXCL|O_CREAT|O_NONBLOCK|O_WRONLY, S_IRUSR|S_IWUSR); if (fd < 0) { - syslog(LOG_CRIT, + EVN_syslog(LOG_CRIT, "cannot open %s: %s\n", - ChrPtr(NW->SpoolFileName), + ChrPtr(NW->tempFileName), strerror(errno)); NW->State = eQUIT - 1; @@ -245,6 +333,9 @@ eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) eNextState NWC_SendREAD(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; + eNextState rc; + if (NW->IO.IOB.TotalSentAlready < NW->IO.IOB.TotalSendSize) { /* @@ -253,88 +344,120 @@ eNextState NWC_SendREAD(AsyncNetworker *NW) */ if (server_shutting_down) { - close(NW->IO.IOB.OtherFD); -////// unlink(ChrPtr(NW->tempFileName)); + FDIOBufferDelete(&NW->IO.IOB); + unlink(ChrPtr(NW->tempFileName)); + FDIOBufferDelete(&IO->IOB); return eAbort; } StrBufPrintf(NW->IO.SendBuf.Buf, "READ %ld|%ld\n", NW->IO.IOB.TotalSentAlready, + NW->IO.IOB.TotalSendSize); +/* ((NW->IO.IOB.TotalSendSize - NW->IO.IOB.TotalSentAlready > IGNET_PACKET_SIZE) ? IGNET_PACKET_SIZE : (NW->IO.IOB.TotalSendSize - NW->IO.IOB.TotalSentAlready)) ); +*/ + NWC_DBG_SEND(); return eSendReply; + } + else + { + NW->State = eCLOS; + rc = NWC_DispatchWriteDone(&NW->IO); + NWC_DBG_SEND(); - - + return rc; } - else {} // continue sending - return eSendReply; } eNextState NWC_ReadREADState(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] == '6') { NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize = atol(ChrPtr(NW->IO.IOBuf)+4); -/// NW->IO.IOB.TotalSentAlready += NW->IO.IOB.ChunkSize; -/// TODO StrBufReadjustIOBuffer(NW->IO.RecvBuf, NW->BlobReadSize); return eReadFile; } + FDIOBufferDelete(&IO->IOB); return eAbort; } eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW); eNextState NWC_ReadREADBlob(AsyncNetworker *NW) { - /// FlushIOBuffer(NW->IO.RecvBuf); /// TODO + eNextState rc; + AsyncIO *IO = &NW->IO; + NWC_DBG_READ(); + if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) + { + NW->State ++; - ///NW->bytes_received += NW->IO.IOB.ChunkSize; + FDIOBufferDelete(&NW->IO.IOB); - if (NW->IO.IOB.TotalSentAlready < NW->IO.IOB.TotalSendSize) - { - NW->State = eREAD - 1; - return eSendReply;/* now fetch next chunk*/ + if (link(ChrPtr(NW->tempFileName), ChrPtr(NW->SpoolFileName)) != 0) { + EVN_syslog(LOG_ALERT, + "Could not link %s to %s: %s\n", + ChrPtr(NW->tempFileName), + ChrPtr(NW->SpoolFileName), + strerror(errno)); + } + + unlink(ChrPtr(NW->tempFileName)); + rc = NWC_DispatchWriteDone(&NW->IO); + NW->State --; + return rc; + } + else { + NW->State --; + NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize; + return eSendReply; //NWC_DispatchWriteDone(&NW->IO); } - else - return NWC_ReadREADBlobDone(NW); } eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW) { - if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) + eNextState rc; + AsyncIO *IO = &NW->IO; +/* we don't have any data to debug print here. */ + if (NW->IO.IOB.TotalSentAlready >= NW->IO.IOB.TotalSendSize) { NW->State ++; - close(NW->IO.IOB.OtherFD); - - if (link(ChrPtr(NW->SpoolFileName), ChrPtr(NW->tempFileName)) != 0) { - syslog(LOG_ALERT, + FDIOBufferDelete(&NW->IO.IOB); + if (link(ChrPtr(NW->tempFileName), ChrPtr(NW->SpoolFileName)) != 0) { + EVN_syslog(LOG_ALERT, "Could not link %s to %s: %s\n", ChrPtr(NW->tempFileName), ChrPtr(NW->SpoolFileName), strerror(errno)); } -///// unlink(ChrPtr(NW->tempFileName)); - return NWC_DispatchWriteDone(&NW->IO); + unlink(ChrPtr(NW->tempFileName)); + rc = NWC_DispatchWriteDone(&NW->IO); + NW->State --; + return rc; } else { NW->State --; + NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize; return NWC_DispatchWriteDone(&NW->IO); } } eNextState NWC_SendCLOS(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; StrBufPlain(NW->IO.SendBuf.Buf, HKEY("CLOS\n")); -//// unlink(ChrPtr(NW->tempFileName)); - return eReadMessage; + NWC_DBG_SEND(); + return eSendReply; } eNextState NWC_ReadCLOSReply(AsyncNetworker *NW) { -/// todo + AsyncIO *IO = &NW->IO; + NWC_DBG_READ(); + FDIOBufferDelete(&IO->IOB); if (ChrPtr(NW->IO.IOBuf)[0] != '2') return eTerminateConnection; return eSendReply; @@ -343,128 +466,149 @@ eNextState NWC_ReadCLOSReply(AsyncNetworker *NW) eNextState NWC_SendNUOP(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; + eNextState rc; long TotalSendSize; struct stat statbuf; int fd; - StrBufPrintf(NW->tempFileName, + StrBufPrintf(NW->SpoolFileName, "%s/%s", ctdl_netout_dir, ChrPtr(NW->node)); - fd = open(ChrPtr(NW->tempFileName), O_RDONLY); + StrBufStripSlashes(NW->SpoolFileName, 1); + + fd = open(ChrPtr(NW->SpoolFileName), O_EXCL|O_NONBLOCK|O_RDONLY); if (fd < 0) { if (errno != ENOENT) { - syslog(LOG_CRIT, + EVN_syslog(LOG_CRIT, "cannot open %s: %s\n", - ChrPtr(NW->tempFileName), + ChrPtr(NW->SpoolFileName), strerror(errno)); } NW->State = eQUIT; - return NWC_SendQUIT(NW); + rc = NWC_SendQUIT(NW); + NWC_DBG_SEND(); + return rc; } if (fstat(fd, &statbuf) == -1) { - syslog(9, "FSTAT FAILED %s [%s]--\n", - ChrPtr(NW->tempFileName), - strerror(errno)); + EVN_syslog(LOG_CRIT, "FSTAT FAILED %s [%s]--\n", + ChrPtr(NW->SpoolFileName), + strerror(errno)); if (fd > 0) close(fd); return eAbort; } TotalSendSize = statbuf.st_size; if (TotalSendSize == 0) { - syslog(LOG_DEBUG, + EVNM_syslog(LOG_DEBUG, "Nothing to send.\n"); NW->State = eQUIT; - return NWC_SendQUIT(NW); + rc = NWC_SendQUIT(NW); + NWC_DBG_SEND(); + if (fd > 0) close(fd); + return rc; } FDIOBufferInit(&NW->IO.IOB, &NW->IO.SendBuf, fd, TotalSendSize); -//// NW->bytes_written = 0; - StrBufPlain(NW->IO.SendBuf.Buf, HKEY("NUOP\n")); + NWC_DBG_SEND(); return eSendReply; } eNextState NWC_ReadNUOPReply(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); - if (ChrPtr(NW->IO.IOBuf)[0] != '2') + if (ChrPtr(NW->IO.IOBuf)[0] != '2') { + FDIOBufferDelete(&IO->IOB); return eAbort; + } return eSendReply; } eNextState NWC_SendWRIT(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; StrBufPrintf(NW->IO.SendBuf.Buf, "WRIT %ld\n", NW->IO.IOB.TotalSendSize - NW->IO.IOB.TotalSentAlready); - + NWC_DBG_SEND(); return eSendReply; } eNextState NWC_ReadWRITReply(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] != '7') { + FDIOBufferDelete(&IO->IOB); return eAbort; } NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize = atol(ChrPtr(NW->IO.IOBuf)+4); -/// NW->IO.IOB.TotalSentAlready += NW->IO.IOB.ChunkSize; return eSendFile; } eNextState NWC_SendBlobDone(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; eNextState rc; - if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) + if (NW->IO.IOB.TotalSentAlready >= IO->IOB.TotalSendSize) { NW->State ++; - close(NW->IO.IOB.OtherFD); -//// TODO: unlink networker file? - rc = NWC_DispatchWriteDone(&NW->IO); + FDIOBufferDelete(&IO->IOB); + rc = NWC_DispatchWriteDone(IO); NW->State --; return rc; } else { NW->State --; - return NWC_DispatchWriteDone(&NW->IO); + IO->IOB.ChunkSendRemain = IO->IOB.ChunkSize; + rc = NWC_DispatchWriteDone(IO); + NW->State --; + return rc; } } eNextState NWC_SendUCLS(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; StrBufPlain(NW->IO.SendBuf.Buf, HKEY("UCLS 1\n")); + NWC_DBG_SEND(); return eSendReply; } eNextState NWC_ReadUCLS(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); - syslog(LOG_NOTICE, "Sent %ld octets to <%s>\n", NW->IO.IOB.ChunkSize, ChrPtr(NW->node)); -/// syslog(LOG_DEBUG, "<%s\n", buf); + EVN_syslog(LOG_NOTICE, "Sent %ld octets to <%s>\n", NW->IO.IOB.ChunkSize, ChrPtr(NW->node)); if (ChrPtr(NW->IO.IOBuf)[0] == '2') { - syslog(LOG_DEBUG, "Removing <%s>\n", ChrPtr(NW->tempFileName)); -/// unlink(ChrPtr(NW->tempFileName)); + EVN_syslog(LOG_DEBUG, "Removing <%s>\n", ChrPtr(NW->SpoolFileName)); + unlink(ChrPtr(NW->SpoolFileName)); } + FDIOBufferDelete(&IO->IOB); return eSendReply; } eNextState NWC_SendQUIT(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; StrBufPlain(NW->IO.SendBuf.Buf, HKEY("QUIT\n")); - network_talking_to(ChrPtr(NW->node), NTT_REMOVE); + NWC_DBG_SEND(); return eSendReply; } eNextState NWC_ReadQUIT(AsyncNetworker *NW) { + AsyncIO *IO = &NW->IO; NWC_DBG_READ(); - return eTerminateConnection; + return eAbort; } @@ -479,8 +623,9 @@ NWClientHandler NWC_ReadHandlers[] = { NWC_ReadWRITReply, NWC_SendBlobDone, NWC_ReadUCLS, - NWC_ReadQUIT -}; + NWC_ReadQUIT}; + +long NWC_ConnTimeout = 100; const long NWC_SendTimeouts[] = { 100, @@ -575,22 +720,17 @@ eNextState nwc_get_one_host_ip(AsyncIO *IO) { AsyncNetworker *NW = IO->Data; /* - * here we start with the lookup of one host. it might be... - * - the relay host *sigh* - * - the direct hostname if there was no mx record - * - one of the mx'es + * here we start with the lookup of one host. */ - InitC_ares_dns(IO); + EVN_syslog(LOG_DEBUG, "NWC: %s\n", __FUNCTION__); - syslog(LOG_DEBUG, "NWC: %s\n", __FUNCTION__); - - syslog(LOG_DEBUG, - "NWC client[%ld]: looking up %s-Record %s : %d ...\n", - NW->n, - (NW->IO.ConnectMe->IPv6)? "aaaa": "a", - NW->IO.ConnectMe->Host, - NW->IO.ConnectMe->Port); + EVN_syslog(LOG_DEBUG, + "NWC client[%ld]: looking up %s-Record %s : %d ...\n", + NW->n, + (NW->IO.ConnectMe->IPv6)? "aaaa": "a", + NW->IO.ConnectMe->Host, + NW->IO.ConnectMe->Port); QueueQuery((NW->IO.ConnectMe->IPv6)? ns_t_aaaa : ns_t_a, NW->IO.ConnectMe->Host, @@ -605,7 +745,7 @@ eNextState nwc_get_one_host_ip(AsyncIO *IO) */ eReadState NWC_ReadServerStatus(AsyncIO *IO) { - AsyncNetworker *NW = IO->Data; +// AsyncNetworker *NW = IO->Data; eReadState Finished = eBufferNotEmpty; switch (IO->NextState) { @@ -626,7 +766,6 @@ eReadState NWC_ReadServerStatus(AsyncIO *IO) case eReadFile: case eSendFile: case eReadPayload: -////TODO Finished = IOBufferStrLength(&IO->RecvBuf) >= NW->BlobReadSize; break; } return Finished; @@ -636,30 +775,71 @@ eReadState NWC_ReadServerStatus(AsyncIO *IO) eNextState NWC_FailNetworkConnection(AsyncIO *IO) { - return eTerminateConnection; + StopClientWatchers(IO, 1); + return QueueDBOperation(IO, SendFailureMessage); +} + +void NWC_SetTimeout(eNextState NextTCPState, AsyncNetworker *NW) +{ + AsyncIO *IO = &NW->IO; + double Timeout = 0.0; + + EVN_syslog(LOG_DEBUG, "%s - %d\n", __FUNCTION__, NextTCPState); + + switch (NextTCPState) { + case eSendReply: + case eSendMore: + break; + case eReadFile: + case eReadMessage: + Timeout = NWC_ReadTimeouts[NW->State]; + break; + case eReadPayload: + Timeout = 100000; + /* TODO!!! */ + break; + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: + case eSendFile: +//TODO + case eTerminateConnection: + case eDBQuery: + case eAbort: + case eReadMore://// TODO + return; + } + if (Timeout > 0) { + EVN_syslog(LOG_DEBUG, + "%s - %d %f\n", + __FUNCTION__, + NextTCPState, + Timeout); + SetNextTimeout(&NW->IO, Timeout*100); + } } eNextState NWC_DispatchReadDone(AsyncIO *IO) { - syslog(LOG_DEBUG, "NWC: %s\n", __FUNCTION__); + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); AsyncNetworker *NW = IO->Data; eNextState rc; rc = NWC_ReadHandlers[NW->State](NW); if (rc != eReadMore) NW->State++; - ////NWCSetTimeout(rc, NW); + NWC_SetTimeout(rc, NW); return rc; } eNextState NWC_DispatchWriteDone(AsyncIO *IO) { - syslog(LOG_DEBUG, "NWC: %s\n", __FUNCTION__); + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); AsyncNetworker *NW = IO->Data; eNextState rc; rc = NWC_SendHandlers[NW->State](NW); - ////NWCSetTimeout(rc, NW); + NWC_SetTimeout(rc, NW); return rc; } @@ -668,43 +848,55 @@ eNextState NWC_DispatchWriteDone(AsyncIO *IO) /*****************************************************************************/ eNextState NWC_Terminate(AsyncIO *IO) { - syslog(LOG_DEBUG, "Nw: %s\n", __FUNCTION__); -/// FinalizeNetworker(IO); TODO + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + FinalizeNetworker(IO); + return eAbort; +} + +eNextState NWC_TerminateDB(AsyncIO *IO) +{ + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + FinalizeNetworker(IO); return eAbort; } eNextState NWC_Timeout(AsyncIO *IO) { -// AsyncNetworker *NW = IO->Data; + AsyncNetworker *NW = IO->Data; + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); -// StrBufPlain(IO->ErrMsg, CKEY(POP3C_ReadErrors[pMsg->State])); todo + if (NW->IO.ErrMsg == NULL) + NW->IO.ErrMsg = NewStrBuf(); + StrBufPrintf(NW->IO.ErrMsg, "Timeout while talking to %s \r\n", ChrPtr(NW->host)); return NWC_FailNetworkConnection(IO); } eNextState NWC_ConnFail(AsyncIO *IO) { -/// AsyncNetworker *NW = IO->Data; + AsyncNetworker *NW = IO->Data; + + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + if (NW->IO.ErrMsg == NULL) + NW->IO.ErrMsg = NewStrBuf(); + StrBufPrintf(NW->IO.ErrMsg, "failed to connect %s \r\n", ChrPtr(NW->host)); - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); -//// StrBufPlain(IO->ErrMsg, CKEY(POP3C_ReadErrors[pMsg->State])); todo return NWC_FailNetworkConnection(IO); } eNextState NWC_DNSFail(AsyncIO *IO) { -/// AsyncNetworker *NW = IO->Data; + AsyncNetworker *NW = IO->Data; + + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + if (NW->IO.ErrMsg == NULL) + NW->IO.ErrMsg = NewStrBuf(); + StrBufPrintf(NW->IO.ErrMsg, "failed to look up %s \r\n", ChrPtr(NW->host)); - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); -//// StrBufPlain(IO->ErrMsg, CKEY(POP3C_ReadErrors[pMsg->State])); todo return NWC_FailNetworkConnection(IO); } eNextState NWC_Shutdown(AsyncIO *IO) { - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); -//// pop3aggr *pMsg = IO->Data; + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); - ////pMsg->MyQEntry->Status = 3; - ///StrBufPlain(pMsg->MyQEntry->StatusMessage, HKEY("server shutdown during message retrieval.")); -/// FinalizePOP3AggrRun(IO); todo + FinalizeNetworker(IO); return eAbort; } @@ -713,48 +905,42 @@ eNextState nwc_connect_ip(AsyncIO *IO) { AsyncNetworker *NW = IO->Data; - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); - syslog(LOG_DEBUG, "network: polling <%s>\n", ChrPtr(NW->node)); - syslog(LOG_NOTICE, "Connecting to <%s> at %s:%s\n", - ChrPtr(NW->node), - ChrPtr(NW->host), - ChrPtr(NW->port)); + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + EVN_syslog(LOG_NOTICE, "Connecting to <%s> at %s:%s\n", + ChrPtr(NW->node), + ChrPtr(NW->host), + ChrPtr(NW->port)); -//// IO->ConnectMe = &NW->Pop3Host; - /* Bypass the ns lookup result like this: IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1"); */ - - /////// SetConnectStatus(IO); - - return InitEventIO(IO, NW, 100, 100, 1); /* - NWC_ConnTimeout, - NWC_ReadTimeouts[0], - 1);*/ + return EvConnectSock(IO, + NWC_ConnTimeout, + NWC_ReadTimeouts[0], + 1); } +static int NetworkerCount = 0; void RunNetworker(AsyncNetworker *NW) { - CitContext *SubC; - + NW->n = NetworkerCount++; + network_talking_to(SKEY(NW->node), NTT_ADD); + syslog(LOG_DEBUG, "NW[%s][%ld]: polling\n", ChrPtr(NW->node), NW->n); ParseURL(&NW->IO.ConnectMe, NW->Url, 504); - NW->IO.Data = NW; - NW->IO.SendDone = NWC_DispatchWriteDone; - NW->IO.ReadDone = NWC_DispatchReadDone; - NW->IO.Terminate = NWC_Terminate; - NW->IO.LineReader = NWC_ReadServerStatus; - NW->IO.ConnFail = NWC_ConnFail; - NW->IO.DNSFail = NWC_DNSFail; - NW->IO.Timeout = NWC_Timeout; - NW->IO.ShutdownAbort = NWC_Shutdown; - - NW->IO.SendBuf.Buf = NewStrBufPlain(NULL, 1024); - NW->IO.RecvBuf.Buf = NewStrBufPlain(NULL, 1024); - NW->IO.IOBuf = NewStrBuf(); - - NW->IO.NextState = eReadMessage; - SubC = CloneContext (&networker_client_CC); - SubC->session_specific_data = (char*) NW; - NW->IO.CitContext = SubC; + InitIOStruct(&NW->IO, + NW, + eReadMessage, + NWC_ReadServerStatus, + NWC_DNSFail, + NWC_DispatchWriteDone, + NWC_DispatchReadDone, + NWC_Terminate, + NWC_TerminateDB, + NWC_ConnFail, + NWC_Timeout, + NWC_Shutdown); + + safestrncpy(((CitContext *)NW->IO.CitContext)->cs_host, + ChrPtr(NW->host), + sizeof(((CitContext *)NW->IO.CitContext)->cs_host)); if (NW->IO.ConnectMe->IsIP) { QueueEventContext(&NW->IO, @@ -771,87 +957,84 @@ void RunNetworker(AsyncNetworker *NW) * Set "full" to nonzero to force a poll of every node, or to zero to poll * only nodes to which we have data to send. */ -void network_poll_other_citadel_nodes(int full_poll, char *working_ignetcfg) +void network_poll_other_citadel_nodes(int full_poll, HashList *ignetcfg) { + const char *key; + long len; + HashPos *Pos; + void *vCfg; AsyncNetworker *NW; - StrBuf *CfgData; - StrBuf *Line; - const char *lptr; - const char *CfgPtr; - int Done; + StrBuf *SpoolFileName; int poll = 0; - if ((working_ignetcfg == NULL) || (*working_ignetcfg == '\0')) { + if (GetCount(ignetcfg) ==0) { syslog(LOG_DEBUG, "network: no neighbor nodes are configured - not polling.\n"); return; } - CfgData = NewStrBufPlain(working_ignetcfg, -1); - Line = NewStrBufPlain(NULL, StrLength(CfgData)); - Done = 0; - CfgPtr = NULL; - while (!Done) + become_session(&networker_client_CC); + + SpoolFileName = NewStrBufPlain(ctdl_netout_dir, -1); + + Pos = GetNewHashPos(ignetcfg, 0); + + while (GetNextHashPos(ignetcfg, Pos, &len, &key, &vCfg)) { /* Use the string tokenizer to grab one line at a time */ - StrBufSipLine(Line, CfgData, &CfgPtr); - Done = CfgPtr == StrBufNOTNULL; - if (StrLength(Line) > 0) + if(server_shutting_down) + return;/* TODO free stuff*/ + NodeConf *pNode = (NodeConf*) vCfg; + poll = 0; + NW = (AsyncNetworker*)malloc(sizeof(AsyncNetworker)); + memset(NW, 0, sizeof(AsyncNetworker)); + + NW->node = NewStrBufDup(pNode->NodeName); + NW->host = NewStrBufDup(pNode->Host); + NW->port = NewStrBufDup(pNode->Port); + NW->secret = NewStrBufDup(pNode->Secret); + + if ( (StrLength(NW->node) != 0) && + (StrLength(NW->secret) != 0) && + (StrLength(NW->host) != 0) && + (StrLength(NW->port) != 0)) { - if(server_shutting_down) - return;/* TODO free stuff*/ - lptr = NULL; - poll = 0; - NW = (AsyncNetworker*)malloc(sizeof(AsyncNetworker)); - memset(NW, 0, sizeof(AsyncNetworker)); - - NW->node = NewStrBufPlain(NULL, StrLength(Line)); - NW->host = NewStrBufPlain(NULL, StrLength(Line)); - NW->port = NewStrBufPlain(NULL, StrLength(Line)); - NW->secret = NewStrBufPlain(NULL, StrLength(Line)); - - StrBufExtract_NextToken(NW->node, Line, &lptr, '|'); - StrBufExtract_NextToken(NW->secret, Line, &lptr, '|'); - StrBufExtract_NextToken(NW->host, Line, &lptr, '|'); - StrBufExtract_NextToken(NW->port, Line, &lptr, '|'); - if ( (StrLength(NW->node) != 0) && - (StrLength(NW->secret) != 0) && - (StrLength(NW->host) != 0) && - (StrLength(NW->port) != 0)) + poll = full_poll; + if (poll == 0) { - poll = full_poll; - if (poll == 0) - { - NW->SpoolFileName = NewStrBufPlain(ctdl_netout_dir, -1); - StrBufAppendBufPlain(NW->SpoolFileName, HKEY("/"), 0); - StrBufAppendBuf(NW->SpoolFileName, NW->node, 0); - if (access(ChrPtr(NW->SpoolFileName), R_OK) == 0) { - poll = 1; - } + StrBufAppendBufPlain(SpoolFileName, HKEY("/"), 0); + StrBufAppendBuf(SpoolFileName, NW->node, 0); + StrBufStripSlashes(SpoolFileName, 1); + + if (access(ChrPtr(SpoolFileName), R_OK) == 0) { + poll = 1; } } - if (poll) { - NW->Url = NewStrBufPlain(NULL, StrLength(Line)); - StrBufPrintf(NW->Url, "citadel://:%s@%s:%s", - ChrPtr(NW->secret), - ChrPtr(NW->host), - ChrPtr(NW->port)); - if (!network_talking_to(ChrPtr(NW->node), NTT_CHECK)) - { - network_talking_to(ChrPtr(NW->node), NTT_ADD); - RunNetworker(NW); - continue; - } + } + if (poll && + (StrLength(NW->host) > 0) && + strcmp("0.0.0.0", ChrPtr(NW->host))) + { + NW->Url = NewStrBuf(); + StrBufPrintf(NW->Url, "citadel://%s@%s:%s", + ChrPtr(NW->secret), + ChrPtr(NW->host), + ChrPtr(NW->port)); + if (!network_talking_to(SKEY(NW->node), NTT_CHECK)) + { + RunNetworker(NW); + continue; } - DestroyNetworker(NW); } + DeleteNetworker(NW); } - + FreeStrBuf(&SpoolFileName); + DeleteHashPos(&Pos); } void network_do_clientqueue(void) { - char *working_ignetcfg; + HashList *working_ignetcfg; int full_processing = 1; static time_t last_run = 0L; @@ -866,19 +1049,20 @@ void network_do_clientqueue(void) ); } - working_ignetcfg = load_working_ignetcfg(); + working_ignetcfg = load_ignetcfg(); /* * Poll other Citadel nodes. Maybe. If "full_processing" is set * then we poll everyone. Otherwise we only poll nodes we have stuff * to send to. */ network_poll_other_citadel_nodes(full_processing, working_ignetcfg); - if (working_ignetcfg) - free(working_ignetcfg); + DeleteHash(&working_ignetcfg); } - - +void LogDebugEnableNetworkClient(const int n) +{ + NetworkClientDebugEnabled = n; +} /* * Module entry point */ @@ -888,7 +1072,9 @@ CTDL_MODULE_INIT(network_client) { CtdlFillSystemContext(&networker_client_CC, "CitNetworker"); - CtdlRegisterSessionHook(network_do_clientqueue, EVT_TIMER); + CtdlRegisterSessionHook(network_do_clientqueue, EVT_TIMER, PRIO_SEND + 10); + CtdlRegisterDebugFlagHook(HKEY("networkclient"), LogDebugEnableNetworkClient, &NetworkClientDebugEnabled); + } - return "network_client"; + return "networkclient"; }