2008-11-05 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

Added the ability to pool proxy connection.
	The conneciton in FTP with proxy-method=GET is not pooled.
	Proxy-Connection header will not be sent when sending CONNECT
	method.
	* src/DownloadEngine.cc
	* src/DownloadEngine.h
	* src/FtpFinishDownloadCommand.cc
	* src/FtpInitiateConnectionCommand.cc
	* src/FtpNegotiationCommand.cc
	* src/HttpDownloadCommand.cc
	* src/HttpInitiateConnectionCommand.cc
	* src/HttpRequest.cc
	* src/HttpRequest.h
	* src/HttpResponse.cc
	* src/HttpSkipResponseCommand.cc
	* test/HttpRequestTest.cc
	* test/HttpResponseTest.cc

ChangeLog

@@ -1,3 +1,22 @@
+2008-11-05  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
+
+	Added the ability to pool proxy connection.
+	The conneciton in FTP with proxy-method=GET is not pooled.
+	Proxy-Connection header will not be sent when sending CONNECT method.
+	* src/DownloadEngine.cc
+	* src/DownloadEngine.h
+	* src/FtpFinishDownloadCommand.cc
+	* src/FtpInitiateConnectionCommand.cc
+	* src/FtpNegotiationCommand.cc
+	* src/HttpDownloadCommand.cc
+	* src/HttpInitiateConnectionCommand.cc
+	* src/HttpRequest.cc
+	* src/HttpRequest.h
+	* src/HttpResponse.cc
+	* src/HttpSkipResponseCommand.cc
+	* test/HttpRequestTest.cc
+	* test/HttpResponseTest.cc
+	
 2008-11-05  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
 
 	Handle date before epoch.

src/DownloadEngine.cc

@@ -65,6 +65,7 @@
 #include "DNSCache.h"
 #include "AuthConfigFactory.h"
 #include "AuthConfig.h"
+#include "Request.h"
 
 #include "BtRegistry.h"
 #include "BtContext.h"
@@ -955,6 +956,38 @@ void DownloadEngine::poolSocket
   poolSocket(ipaddr, port, e);
 }
 
+void DownloadEngine::poolSocket(const SharedHandle<Request>& request,
+				bool proxyDefined,
+				const SharedHandle<SocketCore>& socket,
+				time_t timeout)
+{
+  if(proxyDefined) {
+    // If proxy is defined, then pool socket with its hostname.
+    poolSocket(request->getHost(), request->getPort(), socket);
+  } else {
+    std::pair<std::string, uint16_t> peerInfo;
+    socket->getPeerInfo(peerInfo);
+    poolSocket(peerInfo.first, peerInfo.second, socket);
+  }
+}
+
+void DownloadEngine::poolSocket
+(const SharedHandle<Request>& request,
+ bool proxyDefined,
+ const SharedHandle<SocketCore>& socket,
+ const std::map<std::string, std::string>& options,				
+ time_t timeout)
+{
+  if(proxyDefined) {
+    // If proxy is defined, then pool socket with its hostname.
+    poolSocket(request->getHost(), request->getPort(), socket, options);
+  } else {
+    std::pair<std::string, uint16_t> peerInfo;
+    socket->getPeerInfo(peerInfo);
+    poolSocket(peerInfo.first, peerInfo.second, socket, options);
+  }
+}
+
 std::multimap<std::string, DownloadEngine::SocketPoolEntry>::iterator
 DownloadEngine::findSocketPoolEntry(const std::string& ipaddr, uint16_t port)
 {

src/DownloadEngine.h

@@ -403,6 +403,17 @@ public:
 		  const SharedHandle<SocketCore>& sock,
 		  time_t timeout = 15);
   
+  void poolSocket(const SharedHandle<Request>& request,
+		  bool proxyDefined,
+		  const SharedHandle<SocketCore>& socket,
+		  const std::map<std::string, std::string>& options,
+		  time_t timeout = 15);
+    
+  void poolSocket(const SharedHandle<Request>& request,
+		  bool proxyDefined,
+		  const SharedHandle<SocketCore>& socket,
+		  time_t timeout = 15);
+
   SharedHandle<SocketCore> popPooledSocket(const std::string& ipaddr,
 					   uint16_t port);
 

src/FtpFinishDownloadCommand.cc

@@ -84,12 +84,10 @@ bool FtpFinishDownloadCommand::execute()
     if(status != 226) {
       throw DlAbortEx(StringFormat(EX_BAD_STATUS, status).str());
     }
-    if(!isProxyDefined() && e->option->getAsBool(PREF_FTP_REUSE_CONNECTION)) {
-      std::pair<std::string, uint16_t> peerInfo;
-      socket->getPeerInfo(peerInfo);
+    if(e->option->getAsBool(PREF_FTP_REUSE_CONNECTION)) {
       std::map<std::string, std::string> options;
       options["baseWorkingDir"] = _ftpConnection->getBaseWorkingDir();
-      e->poolSocket(peerInfo.first, peerInfo.second, socket, options);
+      e->poolSocket(req, isProxyDefined(), socket, options);
     }
   } catch(RecoverableException& e) {
     logger->info(EX_EXCEPTION_CAUGHT, e);

src/FtpInitiateConnectionCommand.cc

@@ -68,26 +68,49 @@ Command* FtpInitiateConnectionCommand::createNextCommand
 {
   Command* command;
   if(!proxyRequest.isNull()) {
-    logger->info(MSG_CONNECTING_TO_SERVER, cuid,
-		 proxyRequest->getHost().c_str(), proxyRequest->getPort());
-    socket.reset(new SocketCore());
-    socket->establishConnection(resolvedAddresses.front(),
-				proxyRequest->getPort());
+    std::map<std::string, std::string> options;
+    SharedHandle<SocketCore> pooledSocket =
+      e->popPooledSocket(options, req->getHost(), req->getPort());
+    if(pooledSocket.isNull()) {
+      logger->info(MSG_CONNECTING_TO_SERVER, cuid,
+		   proxyRequest->getHost().c_str(), proxyRequest->getPort());
+      socket.reset(new SocketCore());
+      socket->establishConnection(resolvedAddresses.front(),
+				  proxyRequest->getPort());
       
-    if(e->option->get(PREF_PROXY_METHOD) == V_GET) {
-      SharedHandle<HttpConnection> hc
-	(new HttpConnection(cuid, socket, e->option));
+      if(e->option->get(PREF_PROXY_METHOD) == V_GET) {
+	SharedHandle<HttpConnection> hc
+	  (new HttpConnection(cuid, socket, e->option));
 	
-      HttpRequestCommand* c =
-	new HttpRequestCommand(cuid, req, _requestGroup, hc, e, socket);
-      c->setProxyRequest(proxyRequest);
-      command = c;
-    } else if(e->option->get(PREF_PROXY_METHOD) == V_TUNNEL) {
-      command = new FtpTunnelRequestCommand(cuid, req, _requestGroup, e,
-					    proxyRequest, socket);
+	HttpRequestCommand* c =
+	  new HttpRequestCommand(cuid, req, _requestGroup, hc, e, socket);
+	c->setProxyRequest(proxyRequest);
+	command = c;
+      } else if(e->option->get(PREF_PROXY_METHOD) == V_TUNNEL) {
+	command = new FtpTunnelRequestCommand(cuid, req, _requestGroup, e,
+					      proxyRequest, socket);
+      } else {
+	// TODO
+	throw DlAbortEx("ERROR");
+      }
     } else {
-      // TODO
-      throw DlAbortEx("ERROR");
+      if(e->option->get(PREF_PROXY_METHOD) == V_TUNNEL) {
+	command =
+	  new FtpNegotiationCommand(cuid, req, _requestGroup, e, pooledSocket,
+				    FtpNegotiationCommand::SEQ_SEND_CWD,
+				    options["baseWorkingDir"]);
+      } else if(e->option->get(PREF_PROXY_METHOD) == V_GET) {
+	SharedHandle<HttpConnection> hc
+	  (new HttpConnection(cuid, pooledSocket, e->option));
+	
+	HttpRequestCommand* c =
+	  new HttpRequestCommand(cuid, req, _requestGroup, hc, e, pooledSocket);
+	c->setProxyRequest(proxyRequest);
+	command = c;
+      } else {
+	// TODO
+	throw DlAbortEx("ERROR");
+      }
     }
   } else {
     std::map<std::string, std::string> options;

src/FtpNegotiationCommand.cc

@@ -618,12 +618,10 @@ bool FtpNegotiationCommand::processSequence(const SegmentHandle& segment) {
 
 void FtpNegotiationCommand::poolConnection() const
 {
-  if(!isProxyDefined() && e->option->getAsBool(PREF_FTP_REUSE_CONNECTION)) {
-    std::pair<std::string, uint16_t> peerInfo;
-    socket->getPeerInfo(peerInfo);
+  if(e->option->getAsBool(PREF_FTP_REUSE_CONNECTION)) {
     std::map<std::string, std::string> options;
     options["baseWorkingDir"] = ftp->getBaseWorkingDir();
-    e->poolSocket(peerInfo.first, peerInfo.second, socket, options);
+    e->poolSocket(req, isProxyDefined(),  socket, options);
   }
 }
 

src/HttpDownloadCommand.cc

@@ -63,17 +63,13 @@ bool HttpDownloadCommand::prepareForNextSegment() {
     e->commands.push_back(command);
     return true;
   } else {
-    if(!isProxyDefined()) {
-      if(req->isPipeliningEnabled() ||
-	 (req->isKeepAliveEnabled() &&
-	  ((!_transferEncodingDecoder.isNull() &&
-	    _requestGroup->downloadFinished()) ||
-	   (uint64_t)_segments.front()->getPositionToWrite() ==
-	   _requestGroup->getTotalLength()))) {
-	std::pair<std::string, uint16_t> peerInfo;
-	socket->getPeerInfo(peerInfo);
-	e->poolSocket(peerInfo.first, peerInfo.second, socket);
-      }
+    if(req->isPipeliningEnabled() ||
+       (req->isKeepAliveEnabled() &&
+	((!_transferEncodingDecoder.isNull() &&
+	  _requestGroup->downloadFinished()) ||
+	 (uint64_t)_segments.front()->getPositionToWrite() ==
+	 _requestGroup->getTotalLength()))) {
+      e->poolSocket(req, isProxyDefined(), socket);
     }
 
     return DownloadCommand::prepareForNextSegment();

src/HttpInitiateConnectionCommand.cc

@@ -65,23 +65,40 @@ Command* HttpInitiateConnectionCommand::createNextCommand
 {
   Command* command;
   if(!proxyRequest.isNull()) {
-    logger->info(MSG_CONNECTING_TO_SERVER, cuid,
-		 proxyRequest->getHost().c_str(), proxyRequest->getPort());
-    socket.reset(new SocketCore());
-    socket->establishConnection(resolvedAddresses.front(),
-				proxyRequest->getPort());
-    if(useProxyTunnel()) {
-      command = new HttpProxyRequestCommand(cuid, req, _requestGroup, e,
-					    proxyRequest, socket);
-    } else if(useProxyGet()) {
-      SharedHandle<HttpConnection> httpConnection(new HttpConnection(cuid, socket, e->option));
-      HttpRequestCommand* c = new HttpRequestCommand(cuid, req, _requestGroup,
-						     httpConnection, e, socket);
-      c->setProxyRequest(proxyRequest);
-      command = c;
+    SharedHandle<SocketCore> pooledSocket =
+      e->popPooledSocket(req->getHost(), req->getPort());
+    if(pooledSocket.isNull()) {
+      logger->info(MSG_CONNECTING_TO_SERVER, cuid,
+		   proxyRequest->getHost().c_str(), proxyRequest->getPort());
+      socket.reset(new SocketCore());
+      socket->establishConnection(resolvedAddresses.front(),
+				  proxyRequest->getPort());
+
+      if(useProxyTunnel()) {
+	command = new HttpProxyRequestCommand(cuid, req, _requestGroup, e,
+					      proxyRequest, socket);
+      } else if(useProxyGet()) {
+	SharedHandle<HttpConnection> httpConnection
+	  (new HttpConnection(cuid, socket, e->option));
+	HttpRequestCommand* c = new HttpRequestCommand(cuid, req, _requestGroup,
+						       httpConnection, e,
+						       socket);
+	c->setProxyRequest(proxyRequest);
+	command = c;
+      } else {
+	// TODO
+	throw DlAbortEx("ERROR");
+      }
     } else {
-      // TODO
-      throw DlAbortEx("ERROR");
+      SharedHandle<HttpConnection> httpConnection
+	(new HttpConnection(cuid, pooledSocket, e->option));
+      HttpRequestCommand* c = new HttpRequestCommand(cuid, req, _requestGroup,
+						     httpConnection, e,
+						     pooledSocket);
+      if(useProxyGet()) {
+	c->setProxyRequest(proxyRequest);
+      }
+      command = c;
     }
   } else {
     SharedHandle<SocketCore> pooledSocket =

src/HttpRequest.cc

@@ -245,11 +245,12 @@ std::string HttpRequest::createProxyRequest() const
     std::string(" HTTP/1.1\r\n")+
     "User-Agent: "+userAgent+"\r\n"+
     "Host: "+getHost()+":"+Util::uitos(getPort())+"\r\n";
-  if(request->isKeepAliveEnabled() || request->isPipeliningEnabled()) {
-    requestLine += "Proxy-Connection: Keep-Alive\r\n";
-  }else {
-    requestLine += "Proxy-Connection: close\r\n";
-  }
+  // TODO Is "Proxy-Connection" needed here?
+//   if(request->isKeepAliveEnabled() || request->isPipeliningEnabled()) {
+//     requestLine += "Proxy-Connection: Keep-Alive\r\n";
+//   }else {
+//     requestLine += "Proxy-Connection: close\r\n";
+//   }
   if(!_proxyRequest->getUsername().empty()) {
     requestLine += getProxyAuthString();
   }
@@ -354,4 +355,9 @@ void HttpRequest::setProxyRequest(const SharedHandle<Request>& proxyRequest)
   _proxyRequest = proxyRequest;
 }
 
+bool HttpRequest::isProxyRequestSet() const
+{
+  return !_proxyRequest.isNull();
+}
+
 } // namespace aria2

src/HttpRequest.h

@@ -176,6 +176,12 @@ public:
    * object.
    */
   void setProxyRequest(const SharedHandle<Request>& proxyRequest);
+  
+  /*
+   * Returns true if non-Null proxy request is set by setProxyRequest().
+   * Otherwise, returns false.
+   */
+  bool isProxyRequestSet() const;
 };
 
 typedef SharedHandle<HttpRequest> HttpRequestHandle;

src/HttpResponse.cc

@@ -259,9 +259,17 @@ Time HttpResponse::getLastModifiedTime() const
 
 bool HttpResponse::supportsPersistentConnection() const
 {
-  return Util::toLower(httpHeader->getFirst(HttpHeader::CONNECTION)).
-    find(HttpHeader::CLOSE) == std::string::npos
-    && httpHeader->getVersion() == HttpHeader::HTTP_1_1;
+  std::string connection =
+    Util::toLower(httpHeader->getFirst(HttpHeader::CONNECTION));
+  std::string version = httpHeader->getVersion();
+
+  return
+    connection.find(HttpHeader::CLOSE) == std::string::npos &&
+    (version == HttpHeader::HTTP_1_1 ||
+     connection.find("keep-alive") != std::string::npos) &&
+    (!httpRequest->isProxyRequestSet() ||
+     Util::toLower(httpHeader->getFirst("Proxy-Connection")).find("keep-alive")
+     != std::string::npos);
 }
 
 } // namespace aria2

src/HttpSkipResponseCommand.cc

@@ -137,10 +137,8 @@ bool HttpSkipResponseCommand::executeInternal()
 
 void HttpSkipResponseCommand::poolConnection() const
 {
-  if(!isProxyDefined() && req->supportsPersistentConnection()) {
-    std::pair<std::string, uint16_t> peerInfo;
-    socket->getPeerInfo(peerInfo);
-    e->poolSocket(peerInfo.first, peerInfo.second, socket);
+  if(req->supportsPersistentConnection()) {
+    e->poolSocket(req, isProxyDefined(), socket);
   }
 }
 

test/HttpRequestTest.cc

@@ -471,7 +471,7 @@ void HttpRequestTest::testCreateProxyRequest()
   std::string expectedText = "CONNECT localhost:80 HTTP/1.1\r\n"
     "User-Agent: aria2\r\n"
     "Host: localhost:80\r\n"
-    "Proxy-Connection: close\r\n"
+    //"Proxy-Connection: close\r\n"
     "\r\n";
 
   CPPUNIT_ASSERT_EQUAL(expectedText, httpRequest.createProxyRequest());
@@ -482,7 +482,7 @@ void HttpRequestTest::testCreateProxyRequest()
   expectedText = "CONNECT localhost:80 HTTP/1.1\r\n"
     "User-Agent: aria2\r\n"
     "Host: localhost:80\r\n"
-    "Proxy-Connection: Keep-Alive\r\n"
+    //"Proxy-Connection: Keep-Alive\r\n"
     "\r\n";
 
   CPPUNIT_ASSERT_EQUAL(expectedText, httpRequest.createProxyRequest());
@@ -494,7 +494,7 @@ void HttpRequestTest::testCreateProxyRequest()
   expectedText = "CONNECT localhost:80 HTTP/1.1\r\n"
     "User-Agent: aria2\r\n"
     "Host: localhost:80\r\n"
-    "Proxy-Connection: Keep-Alive\r\n"
+    //"Proxy-Connection: Keep-Alive\r\n"
     "\r\n";
 
   CPPUNIT_ASSERT_EQUAL(expectedText, httpRequest.createProxyRequest());
@@ -506,7 +506,7 @@ void HttpRequestTest::testCreateProxyRequest()
   expectedText = "CONNECT localhost:80 HTTP/1.1\r\n"
     "User-Agent: aria2\r\n"
     "Host: localhost:80\r\n"
-    "Proxy-Connection: Keep-Alive\r\n"
+    //"Proxy-Connection: Keep-Alive\r\n"
     "Proxy-Authorization: Basic YXJpYTJwcm94eXVzZXI6YXJpYTJwcm94eXBhc3N3ZA==\r\n"
     "\r\n";
 
@@ -601,7 +601,7 @@ void HttpRequestTest::testUserAgent()
   std::string expectedTextForProxy = "CONNECT localhost:8080 HTTP/1.1\r\n"
     "User-Agent: aria2 (Linux)\r\n"
     "Host: localhost:8080\r\n"
-    "Proxy-Connection: close\r\n"
+    //"Proxy-Connection: close\r\n"
     "\r\n";
 
   CPPUNIT_ASSERT_EQUAL(expectedTextForProxy, httpRequest.createProxyRequest());

test/HttpResponseTest.cc

@@ -496,19 +496,64 @@ void HttpResponseTest::testSupportsPersistentConnection()
   HttpResponse httpResponse;
   SharedHandle<HttpHeader> httpHeader(new HttpHeader());
   httpResponse.setHttpHeader(httpHeader);
+  SharedHandle<HttpRequest> httpRequest(new HttpRequest());
+  httpResponse.setHttpRequest(httpRequest);
 
   httpHeader->setVersion("HTTP/1.1");
   CPPUNIT_ASSERT(httpResponse.supportsPersistentConnection());
-
-  httpHeader->setVersion("HTTP/1.0");
-  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
-
-  httpHeader->setVersion("HTTP/1.1");
   httpHeader->put("Connection", "close");
   CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
   httpHeader->clearField();
   httpHeader->put("Connection", "keep-alive");
   CPPUNIT_ASSERT(httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+
+  httpHeader->setVersion("HTTP/1.0");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->put("Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Connection", "keep-alive");
+  CPPUNIT_ASSERT(httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+
+  // test proxy connection
+  SharedHandle<Request> proxyRequest(new Request());
+  httpRequest->setProxyRequest(proxyRequest);
+  
+  httpHeader->setVersion("HTTP/1.1");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->put("Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Connection", "keep-alive");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Proxy-Connection", "keep-alive");
+  CPPUNIT_ASSERT(httpResponse.supportsPersistentConnection());
+  httpHeader->put("Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Proxy-Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+
+  httpHeader->setVersion("HTTP/1.0");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->put("Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Connection", "keep-alive");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->put("Proxy-Connection", "keep-alive");
+  CPPUNIT_ASSERT(httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Proxy-Connection", "keep-alive");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
+  httpHeader->put("Proxy-Connection", "close");
+  CPPUNIT_ASSERT(!httpResponse.supportsPersistentConnection());
+  httpHeader->clearField();
 }
 
 } // namespace aria2