X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Fnetwork%2Fserv_networkclient.c;h=63c6e65412e2165b29eb2852dd1f8ace014eb7de;hb=7cced4381b0497cf3d99a489bbb1a4f5375ded32;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..63c6e6541 100644
--- a/citadel/modules/network/serv_networkclient.c
+++ b/citadel/modules/network/serv_networkclient.c
@@ -125,15 +125,36 @@ 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];
@@ -245,6 +266,8 @@ eNextState NWC_ReadNDOPReply(AsyncNetworker *NW)
 
 eNextState NWC_SendREAD(AsyncNetworker *NW)
 {
+	eNextState rc;
+
 	if (NW->IO.IOB.TotalSentAlready < NW->IO.IOB.TotalSendSize)
 	{
 		/*
@@ -253,23 +276,30 @@ 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));
 			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)
@@ -279,8 +309,6 @@ eNextState NWC_ReadREADState(AsyncNetworker *NW)
 	{
 		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;
 	}
 	return eAbort;
@@ -288,26 +316,39 @@ eNextState NWC_ReadREADState(AsyncNetworker *NW)
 eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW);
 eNextState NWC_ReadREADBlob(AsyncNetworker *NW)
 {
-	/// FlushIOBuffer(NW->IO.RecvBuf); /// TODO
-
-	///NW->bytes_received += NW->IO.IOB.ChunkSize;
-
-	if (NW->IO.IOB.TotalSentAlready < NW->IO.IOB.TotalSendSize)
+	NWC_DBG_READ();
+	if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready)
 	{
-		NW->State = eREAD - 1;
-		return eSendReply;/* now fetch next chunk*/
+		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);
 	}
-	else
-		return NWC_ReadREADBlobDone(NW);
 }
 
 eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW)
 {
+	NWC_DBG_READ();
 	if (NW->IO.IOB.TotalSendSize == NW->IO.IOB.TotalSentAlready)
 	{
 		NW->State ++;
 
-		close(NW->IO.IOB.OtherFD);
+		FDIOBufferDelete(&NW->IO.IOB);
 		
 		if (link(ChrPtr(NW->SpoolFileName), ChrPtr(NW->tempFileName)) != 0) {
 			syslog(LOG_ALERT, 
@@ -317,24 +358,25 @@ eNextState NWC_ReadREADBlobDone(AsyncNetworker *NW)
 			       strerror(errno));
 		}
 	
-/////		unlink(ChrPtr(NW->tempFileName));
+		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;
@@ -343,6 +385,7 @@ eNextState NWC_ReadCLOSReply(AsyncNetworker *NW)
 
 eNextState NWC_SendNUOP(AsyncNetworker *NW)
 {
+	eNextState rc;
 	long TotalSendSize;
 	struct stat statbuf;
 	int fd;
@@ -360,7 +403,8 @@ eNextState NWC_SendNUOP(AsyncNetworker *NW)
 			       strerror(errno));
 		}
 		NW->State = eQUIT;
-		return NWC_SendQUIT(NW);
+		rc = NWC_SendQUIT(NW);
+		NWC_DBG_SEND();
 	}
 
 	if (fstat(fd, &statbuf) == -1) {
@@ -375,13 +419,14 @@ eNextState NWC_SendNUOP(AsyncNetworker *NW)
 		syslog(LOG_DEBUG,
 		       "Nothing to send.\n");
 		NW->State = eQUIT;
-		return NWC_SendQUIT(NW);
+		rc = NWC_SendQUIT(NW);
+		NWC_DBG_SEND();
+		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;
 
 }
@@ -397,7 +442,7 @@ eNextState NWC_SendWRIT(AsyncNetworker *NW)
 {
 	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)
@@ -410,7 +455,6 @@ eNextState NWC_ReadWRITReply(AsyncNetworker *NW)
 
 	NW->IO.IOB.ChunkSendRemain = 
 		NW->IO.IOB.ChunkSize = atol(ChrPtr(NW->IO.IOBuf)+4);
-///	NW->IO.IOB.TotalSentAlready += NW->IO.IOB.ChunkSize;
 	return eSendFile;
 }
 
@@ -421,14 +465,14 @@ eNextState NWC_SendBlobDone(AsyncNetworker *NW)
 	{
 		NW->State ++;
 
-		close(NW->IO.IOB.OtherFD);
-//// TODO: unlink networker file?		
+		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);
 	}
 }
@@ -436,6 +480,7 @@ eNextState NWC_SendBlobDone(AsyncNetworker *NW)
 eNextState NWC_SendUCLS(AsyncNetworker *NW)
 {
 	StrBufPlain(NW->IO.SendBuf.Buf, HKEY("UCLS 1\n"));
+	NWC_DBG_SEND();
 	return eSendReply;
 
 }
@@ -444,10 +489,9 @@ eNextState NWC_ReadUCLS(AsyncNetworker *NW)
 	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);
 	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;
 }
@@ -456,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;
 }
 
@@ -464,7 +508,7 @@ eNextState NWC_ReadQUIT(AsyncNetworker *NW)
 {
 	NWC_DBG_READ();
 
-	return eTerminateConnection;
+	return eAbort;
 }
 
 
@@ -479,8 +523,9 @@ NWClientHandler NWC_ReadHandlers[] = {
 	NWC_ReadWRITReply,
 	NWC_SendBlobDone,
 	NWC_ReadUCLS,
-	NWC_ReadQUIT
-};
+	NWC_ReadQUIT};
+
+long NWC_ConnTimeout = 100;
 
 const long NWC_SendTimeouts[] = {
 	100,
@@ -575,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);
@@ -605,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) {
@@ -626,7 +668,6 @@ eReadState NWC_ReadServerStatus(AsyncIO *IO)
 	case eReadFile:
 	case eSendFile:
 	case eReadPayload:
-////TODO		Finished = IOBufferStrLength(&IO->RecvBuf) >= NW->BlobReadSize;
 		break;
 	}
 	return Finished;
@@ -636,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);
 }
 
 
@@ -649,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)
@@ -659,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;
 }
 
@@ -669,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)
@@ -702,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;
 }
 
@@ -720,41 +789,31 @@ 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,
+			     NWC_ConnTimeout,
+			     NWC_ReadTimeouts[0],
+			     1);
 }
 
 void RunNetworker(AsyncNetworker *NW)
 {
-	CitContext *SubC;
-
 	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_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,
@@ -786,6 +845,8 @@ void network_poll_other_citadel_nodes(int full_poll, char *working_ignetcfg)
 		syslog(LOG_DEBUG, "network: no neighbor nodes are configured - not polling.\n");
 		return;
 	}
+	become_session(&networker_client_CC);
+
 	CfgData = NewStrBufPlain(working_ignetcfg, -1);
 	Line = NewStrBufPlain(NULL, StrLength(CfgData));
 	Done = 0;
@@ -835,16 +896,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);
 
 }