X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Fnetwork%2Fserv_networkclient.c;h=d0e9e6d197afd729d7ab38660907affe0e1b971c;hb=cf7cb2463d47a4a9ed36c8d1c13f188418389437;hp=7515c7926d6c02e5b15c10f8cb0c41bf995b05d8;hpb=388abb1a919918487315401f0b06084e00c33a85;p=citadel.git diff --git a/citadel/modules/network/serv_networkclient.c b/citadel/modules/network/serv_networkclient.c index 7515c7926..d0e9e6d19 100644 --- a/citadel/modules/network/serv_networkclient.c +++ b/citadel/modules/network/serv_networkclient.c @@ -107,10 +107,7 @@ typedef struct _async_networker { AsyncIO IO; DNSQueryParts HostLookup; eNWCState State; - int fd; long n; - FILE *TmpFile; - long bytes_received; StrBuf *SpoolFileName; StrBuf *tempFileName; StrBuf *node; @@ -118,11 +115,6 @@ typedef struct _async_networker { StrBuf *port; StrBuf *secret; StrBuf *Url; - - long download_len; - int BlobReadSize; - long bytes_written; - long bytes_to_write; } AsyncNetworker; typedef eNextState(*NWClientHandler)(AsyncNetworker* NW); @@ -131,17 +123,38 @@ eNextState nwc_get_one_host_ip(AsyncIO *IO); 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); + ((struct CitContext*)NW->IO.CitContext)->state = CON_IDLE; + ((struct CitContext*)NW->IO.CitContext)->kill_me = 1; + FreeAsyncIOContents(&NW->IO); + 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_OK (strncasecmp(ChrPtr(NW->IO.IOBuf), "+OK", 3) == 0) +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]; @@ -215,17 +228,23 @@ eNextState NWC_SendNDOP(AsyncNetworker *NW) eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) { + int TotalSendSize; NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] == '2') { - NW->download_len = atol (ChrPtr(NW->IO.IOBuf) + 4); - syslog(LOG_DEBUG, "Expecting to transfer %ld bytes\n", NW->download_len); - if (NW->download_len <= 0) { + + 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); + if (TotalSendSize <= 0) { NW->State = eNUOP - 1; } else { - NW->TmpFile = fopen(ChrPtr(NW->SpoolFileName), "w"); - if (NW->TmpFile == NULL) + int fd; + fd = open(ChrPtr(NW->SpoolFileName), + O_EXCL|O_CREAT|O_NONBLOCK|O_WRONLY, + S_IRUSR|S_IWUSR); + if (fd < 0) { syslog(LOG_CRIT, "cannot open %s: %s\n", @@ -233,7 +252,9 @@ eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) strerror(errno)); NW->State = eQUIT - 1; + return eAbort; } + FDIOBufferInit(&NW->IO.IOB, &NW->IO.RecvBuf, fd, TotalSendSize); } return eSendReply; } @@ -245,7 +266,9 @@ eNextState NWC_ReadNDOPReply(AsyncNetworker *NW) eNextState NWC_SendREAD(AsyncNetworker *NW) { - if (NW->bytes_received < NW->download_len) + eNextState rc; + + if (NW->IO.IOB.TotalSentAlready < NW->IO.IOB.TotalSendSize) { /* * If shutting down we can exit here and unlink the temp file. @@ -253,23 +276,30 @@ eNextState NWC_SendREAD(AsyncNetworker *NW) */ if (server_shutting_down) { - fclose(NW->TmpFile); -////// unlink(ChrPtr(NW->tempFileName)); + FDIOBufferDelete(&NW->IO.IOB); + unlink(ChrPtr(NW->tempFileName)); return eAbort; } StrBufPrintf(NW->IO.SendBuf.Buf, "READ %ld|%ld\n", - NW->bytes_received, - ((NW->download_len - NW->bytes_received > IGNET_PACKET_SIZE) + NW->IO.IOB.TotalSentAlready, + NW->IO.IOB.TotalSendSize); +/* + ((NW->IO.IOB.TotalSendSize - NW->IO.IOB.TotalSentAlready > IGNET_PACKET_SIZE) ? IGNET_PACKET_SIZE : - (NW->download_len - NW->bytes_received)) + (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) @@ -277,28 +307,21 @@ eNextState NWC_ReadREADState(AsyncNetworker *NW) NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] == '6') { - NW->BlobReadSize = atol(ChrPtr(NW->IO.IOBuf)+4); - NW->bytes_received += NW->BlobReadSize; -/// TODO StrBufReadjustIOBuffer(NW->IO.RecvBuf, NW->BlobReadSize); - return eReadPayload; + NW->IO.IOB.ChunkSendRemain = + NW->IO.IOB.ChunkSize = atol(ChrPtr(NW->IO.IOBuf)+4); + return eReadFile; } return eAbort; } +eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW); eNextState NWC_ReadREADBlob(AsyncNetworker *NW) { - /// FlushIOBuffer(NW->IO.RecvBuf); /// TODO - fwrite(NW->IO.RecvBuf.ReadWritePointer, - 1, - NW->BlobReadSize, - NW->TmpFile); - if (NW->bytes_received < NW->download_len) - { - NW->State = eREAD - 1; - return eSendReply;/* now fetch next chunk*/ - } - else + NWC_DBG_READ(); + if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) { - fclose(NW->TmpFile); + NW->State ++; + + FDIOBufferDelete(&NW->IO.IOB); if (link(ChrPtr(NW->SpoolFileName), ChrPtr(NW->tempFileName)) != 0) { syslog(LOG_ALERT, @@ -308,22 +331,52 @@ eNextState NWC_ReadREADBlob(AsyncNetworker *NW) strerror(errno)); } -///// unlink(ChrPtr(NW->tempFileName)); - return eSendReply; //// TODO: step forward. + unlink(ChrPtr(NW->tempFileName)); + return NWC_DispatchWriteDone(&NW->IO); + } + else { + NW->State --; + NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize; + return NWC_DispatchWriteDone(&NW->IO); } } +eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW) +{ + NWC_DBG_READ(); + if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) + { + NW->State ++; + FDIOBufferDelete(&NW->IO.IOB); + + if (link(ChrPtr(NW->SpoolFileName), ChrPtr(NW->tempFileName)) != 0) { + 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); + } + else { + NW->State --; + NW->IO.IOB.ChunkSendRemain = NW->IO.IOB.ChunkSize; + return NWC_DispatchWriteDone(&NW->IO); + } +} eNextState NWC_SendCLOS(AsyncNetworker *NW) { 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 + NWC_DBG_READ(); if (ChrPtr(NW->IO.IOBuf)[0] != '2') return eTerminateConnection; return eSendReply; @@ -332,14 +385,17 @@ eNextState NWC_ReadCLOSReply(AsyncNetworker *NW) eNextState NWC_SendNUOP(AsyncNetworker *NW) { + eNextState rc; + long TotalSendSize; struct stat statbuf; + int fd; StrBufPrintf(NW->tempFileName, "%s/%s", ctdl_netout_dir, ChrPtr(NW->node)); - NW->fd = open(ChrPtr(NW->tempFileName), O_RDONLY); - if (NW->fd < 0) { + fd = open(ChrPtr(NW->tempFileName), O_RDONLY); + if (fd < 0) { if (errno != ENOENT) { syslog(LOG_CRIT, "cannot open %s: %s\n", @@ -347,28 +403,30 @@ eNextState NWC_SendNUOP(AsyncNetworker *NW) strerror(errno)); } NW->State = eQUIT; - return NWC_SendQUIT(NW); + rc = NWC_SendQUIT(NW); + NWC_DBG_SEND(); } - if (fstat(NW->fd, &statbuf) == -1) { + if (fstat(fd, &statbuf) == -1) { syslog(9, "FSTAT FAILED %s [%s]--\n", ChrPtr(NW->tempFileName), strerror(errno)); - if (NW->fd > 0) close(NW->fd); + if (fd > 0) close(fd); return eAbort; } - - NW->download_len = statbuf.st_size; - if (NW->download_len == 0) { + TotalSendSize = statbuf.st_size; + if (TotalSendSize == 0) { syslog(LOG_DEBUG, "Nothing to send.\n"); NW->State = eQUIT; - return NWC_SendQUIT(NW); + rc = NWC_SendQUIT(NW); + NWC_DBG_SEND(); + return rc; } - - NW->bytes_written = 0; + FDIOBufferInit(&NW->IO.IOB, &NW->IO.SendBuf, fd, TotalSendSize); StrBufPlain(NW->IO.SendBuf.Buf, HKEY("NUOP\n")); + NWC_DBG_SEND(); return eSendReply; } @@ -382,8 +440,9 @@ eNextState NWC_ReadNUOPReply(AsyncNetworker *NW) eNextState NWC_SendWRIT(AsyncNetworker *NW) { - StrBufPrintf(NW->IO.SendBuf.Buf, "WRIT %ld\n", NW->bytes_to_write); - + 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) @@ -394,22 +453,34 @@ eNextState NWC_ReadWRITReply(AsyncNetworker *NW) return eAbort; } - NW->BlobReadSize = atol(ChrPtr(NW->IO.IOBuf)+4); - return eSendMore; + NW->IO.IOB.ChunkSendRemain = + NW->IO.IOB.ChunkSize = atol(ChrPtr(NW->IO.IOBuf)+4); + return eSendFile; } -eNextState NWC_SendBlob(AsyncNetworker *NW) +eNextState NWC_SendBlobDone(AsyncNetworker *NW) { + eNextState rc; + if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready) + { + NW->State ++; - /// bytes_to_write -= thisblock; - /// bytes_written += thisblock; - - return eReadMessage; + FDIOBufferDelete(&NW->IO.IOB); + 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_SendUCLS(AsyncNetworker *NW) { StrBufPlain(NW->IO.SendBuf.Buf, HKEY("UCLS 1\n")); + NWC_DBG_SEND(); return eSendReply; } @@ -417,11 +488,10 @@ eNextState NWC_ReadUCLS(AsyncNetworker *NW) { NWC_DBG_READ(); - syslog(LOG_NOTICE, "Sent %ld octets to <%s>\n", NW->bytes_written, ChrPtr(NW->node)); -/// syslog(LOG_DEBUG, "<%s\n", buf); + 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)); + unlink(ChrPtr(NW->tempFileName)); } return eSendReply; } @@ -430,7 +500,7 @@ eNextState NWC_SendQUIT(AsyncNetworker *NW) { StrBufPlain(NW->IO.SendBuf.Buf, HKEY("QUIT\n")); - network_talking_to(ChrPtr(NW->node), NTT_REMOVE); + NWC_DBG_SEND(); return eSendReply; } @@ -438,7 +508,7 @@ eNextState NWC_ReadQUIT(AsyncNetworker *NW) { NWC_DBG_READ(); - return eTerminateConnection; + return eAbort; } @@ -451,10 +521,11 @@ NWClientHandler NWC_ReadHandlers[] = { NWC_ReadCLOSReply, NWC_ReadNUOPReply, NWC_ReadWRITReply, - NULL, + NWC_SendBlobDone, NWC_ReadUCLS, - NWC_ReadQUIT -}; + NWC_ReadQUIT}; + +long NWC_ConnTimeout = 100; const long NWC_SendTimeouts[] = { 100, @@ -482,11 +553,11 @@ NWClientHandler NWC_SendHandlers[] = { NWC_SendAuth, NWC_SendNDOP, NWC_SendREAD, - NULL, + NWC_ReadREADBlobDone, NWC_SendCLOS, NWC_SendNUOP, NWC_SendWRIT, - NWC_SendBlob, + NWC_SendBlobDone, NWC_SendUCLS, NWC_SendQUIT }; @@ -549,10 +620,7 @@ 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); @@ -579,7 +647,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) { @@ -597,8 +665,9 @@ eReadState NWC_ReadServerStatus(AsyncIO *IO) case eReadMessage: Finished = StrBufChunkSipLine(IO->IOBuf, &IO->RecvBuf); break; + case eReadFile: + case eSendFile: case eReadPayload: - Finished = IOBufferStrLength(&IO->RecvBuf) >= NW->BlobReadSize; break; } return Finished; @@ -608,7 +677,39 @@ eReadState NWC_ReadServerStatus(AsyncIO *IO) eNextState NWC_FailNetworkConnection(AsyncIO *IO) { - return eTerminateConnection; + return eAbort; +} + +void NWC_SetTimeout(eNextState NextTCPState, AsyncNetworker *NW) +{ + double Timeout = 0.0; + + syslog(LOG_DEBUG, "NWC3: %s\n", __FUNCTION__); + + 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; + } + SetNextTimeout(&NW->IO, Timeout); } @@ -621,7 +722,7 @@ eNextState NWC_DispatchReadDone(AsyncIO *IO) 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) @@ -631,7 +732,7 @@ eNextState NWC_DispatchWriteDone(AsyncIO *IO) eNextState rc; rc = NWC_SendHandlers[NW->State](NW); - ////NWCSetTimeout(rc, NW); + NWC_SetTimeout(rc, NW); return rc; } @@ -641,16 +742,14 @@ eNextState NWC_DispatchWriteDone(AsyncIO *IO) eNextState NWC_Terminate(AsyncIO *IO) { syslog(LOG_DEBUG, "Nw: %s\n", __FUNCTION__); -/// FinalizeNetworker(IO); TODO + FinalizeNetworker(IO); return eAbort; } eNextState NWC_Timeout(AsyncIO *IO) { -// AsyncNetworker *NW = IO->Data; - syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); -// StrBufPlain(IO->ErrMsg, CKEY(POP3C_ReadErrors[pMsg->State])); todo + return NWC_FailNetworkConnection(IO); } eNextState NWC_ConnFail(AsyncIO *IO) @@ -674,9 +773,7 @@ eNextState NWC_Shutdown(AsyncIO *IO) syslog(LOG_DEBUG, "NW: %s\n", __FUNCTION__); //// pop3aggr *pMsg = IO->Data; - ////pMsg->MyQEntry->Status = 3; - ///StrBufPlain(pMsg->MyQEntry->StatusMessage, HKEY("server shutdown during message retrieval.")); -/// FinalizePOP3AggrRun(IO); todo + FinalizeNetworker(IO); return eAbort; } @@ -692,15 +789,10 @@ eNextState nwc_connect_ip(AsyncIO *IO) 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, NW, + NWC_ConnTimeout, + NWC_ReadTimeouts[0], + 1); } void RunNetworker(AsyncNetworker *NW) @@ -715,7 +807,7 @@ void RunNetworker(AsyncNetworker *NW) NW->IO.Terminate = NWC_Terminate; NW->IO.LineReader = NWC_ReadServerStatus; NW->IO.ConnFail = NWC_ConnFail; - NW->IO.DNSFail = NWC_DNSFail; + NW->IO.DNS.Fail = NWC_DNSFail; NW->IO.Timeout = NWC_Timeout; NW->IO.ShutdownAbort = NWC_Shutdown; @@ -728,6 +820,10 @@ void RunNetworker(AsyncNetworker *NW) SubC->session_specific_data = (char*) NW; NW->IO.CitContext = SubC; + safestrncpy(SubC->cs_host, + ChrPtr(NW->host), + sizeof(SubC->cs_host)); + if (NW->IO.ConnectMe->IsIP) { QueueEventContext(&NW->IO, nwc_connect_ip); @@ -807,16 +903,18 @@ void network_poll_other_citadel_nodes(int full_poll, char *working_ignetcfg) ChrPtr(NW->secret), ChrPtr(NW->host), ChrPtr(NW->port)); - if (!network_talking_to(ChrPtr(NW->node), NTT_CHECK)) + if (!network_talking_to(SKEY(NW->node), NTT_CHECK)) { - network_talking_to(ChrPtr(NW->node), NTT_ADD); + network_talking_to(SKEY(NW->node), NTT_ADD); RunNetworker(NW); continue; } } - DestroyNetworker(NW); + DeleteNetworker(NW); } } + FreeStrBuf(&CfgData); + FreeStrBuf(&Line); }