2010-09-06 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>

	Data from remote server in HTTP/FTP download are now written to
	the disk(or memory) through StreamFilter. Decoding chunked and
	gziped streams are done cascading StreamFilter.
	Removed inefficient 1byte read code.
	* src/ChunkedDecodingStreamFilter.cc
	* src/ChunkedDecodingStreamFilter.h
	* src/DownloadCommand.cc
	* src/DownloadCommand.h
	* src/GZipDecodingStreamFilter.cc
	* src/GZipDecodingStreamFilter.h
	* src/HttpConnection.cc
	* src/HttpDownloadCommand.cc
	* src/HttpResponse.cc
	* src/HttpResponse.h
	* src/HttpResponseCommand.cc
	* src/HttpResponseCommand.h
	* src/HttpSkipResponseCommand.cc
	* src/HttpSkipResponseCommand.h
	* src/Makefile.am
	* src/NullSinkStreamFilter.cc
	* src/NullSinkStreamFilter.h
	* src/RequestGroup.cc
	* src/SinkStreamFilter.cc
	* src/SinkStreamFilter.h
	* src/StreamFilter.cc
	* src/StreamFilter.h
	* test/ChunkedDecodingStreamFilterTest.cc
	* test/GZipDecodingStreamFilterTest.cc
	* test/HttpResponseTest.cc
	* test/Makefile.am
	* test/MockSegment.h

1 files changed, 53 insertions, 87 deletions

diff --git a/src/DownloadCommand.cc b/src/DownloadCommand.cc
index f79a3adc..fb00ef38 100644
--- a/src/DownloadCommand.cc
+++ b/src/DownloadCommand.cc
@@ -56,11 +56,11 @@
 #include "message.h"
 #include "prefs.h"
 #include "StringFormat.h"
-#include "Decoder.h"
 #include "RequestGroupMan.h"
 #include "wallclock.h"
 #include "ServerStatMan.h"
 #include "FileAllocationEntry.h"
+#include "SinkStreamFilter.h"
 #ifdef ENABLE_MESSAGE_DIGEST
 # include "MessageDigestHelper.h"
 #endif // ENABLE_MESSAGE_DIGEST
@@ -104,6 +104,10 @@ DownloadCommand::DownloadCommand(cuid_t cuid,
   peerStat_ = req->initPeerStat();
   peerStat_->downloadStart();
   getSegmentMan()->registerPeerStat(peerStat_);
+
+  streamFilter_.reset(new SinkStreamFilter(pieceHashValidationEnabled_));
+  streamFilter_->init();
+  sinkFilterOnly_ = true;
 }
 
 DownloadCommand::~DownloadCommand() {
@@ -120,80 +124,45 @@ bool DownloadCommand::executeInternal() {
     return false;
   }
   setReadCheckSocket(getSocket());
-  SharedHandle<Segment> segment = getSegments().front();
 
+  const SharedHandle<DiskAdaptor>& diskAdaptor =
+    getPieceStorage()->getDiskAdaptor();
+  SharedHandle<Segment> segment = getSegments().front();
   size_t bufSize;
-  if(segment->getLength() > 0) {
-    if(static_cast<uint64_t>(segment->getPosition()+segment->getLength()) <=
-       static_cast<uint64_t>(getFileEntry()->getLastOffset())) {
-      bufSize = std::min(segment->getLength()-segment->getWrittenLength(),
-                         BUFSIZE);
+  if(sinkFilterOnly_) {
+    if(segment->getLength() > 0 ) {
+      if(static_cast<uint64_t>(segment->getPosition()+segment->getLength()) <=
+         static_cast<uint64_t>(getFileEntry()->getLastOffset())) {
+        bufSize = std::min(segment->getLength()-segment->getWrittenLength(),
+                           BUFSIZE);
+      } else {
+        bufSize =
+          std::min
+          (static_cast<size_t>
+           (getFileEntry()->getLastOffset()-segment->getPositionToWrite()),
+           BUFSIZE);
+      }
     } else {
-      bufSize =
-        std::min
-        (static_cast<size_t>
-         (getFileEntry()->getLastOffset()-segment->getPositionToWrite()),
-         BUFSIZE);
+      bufSize = BUFSIZE;
     }
+    getSocket()->readData(buf_, bufSize);
+    streamFilter_->transform(diskAdaptor, segment, buf_, bufSize);
   } else {
+    // It is possible that segment is completed but we have some bytes
+    // of stream to read. For example, chunked encoding has "0"+CRLF
+    // after data. After we read data(at this moment segment is
+    // completed), we need another 3bytes(or more if it has trailers).
     bufSize = BUFSIZE;
-  }
-  // It is possible that segment is completed but we have some bytes
-  // of stream to read. For example, chunked encoding has "0"+CRLF
-  // after data. After we read data(at this moment segment is
-  // completed), we need another 3bytes(or more if it has extension).
-  if(bufSize == 0 &&
-     ((!transferEncodingDecoder_.isNull() &&
-       !transferEncodingDecoder_->finished()) ||
-      (!contentEncodingDecoder_.isNull() &&
-       !contentEncodingDecoder_->finished()))) {
-    bufSize = 1;
-  }
-  getSocket()->readData(buf_, bufSize);
-
-  const SharedHandle<DiskAdaptor>& diskAdaptor =
-    getPieceStorage()->getDiskAdaptor();
-
-  const unsigned char* bufFinal;
-  size_t bufSizeFinal;
-
-  std::string decoded;
-  if(transferEncodingDecoder_.isNull()) {
-    bufFinal = buf_;
-    bufSizeFinal = bufSize;
-  } else {
-    decoded = transferEncodingDecoder_->decode(buf_, bufSize);
-
-    bufFinal = reinterpret_cast<const unsigned char*>(decoded.c_str());
-    bufSizeFinal = decoded.size();
-  }
-
-  if(contentEncodingDecoder_.isNull()) {
-    diskAdaptor->writeData(bufFinal, bufSizeFinal,
-                           segment->getPositionToWrite());
-  } else {
-    std::string out = contentEncodingDecoder_->decode(bufFinal, bufSizeFinal);
-    diskAdaptor->writeData(reinterpret_cast<const unsigned char*>(out.data()),
-                           out.size(),
-                           segment->getPositionToWrite());
-    bufSizeFinal = out.size();
-  }
-
-#ifdef ENABLE_MESSAGE_DIGEST
-
-  if(pieceHashValidationEnabled_) {
-    segment->updateHash(segment->getWrittenLength(), bufFinal, bufSizeFinal);
-  }
-
-#endif // ENABLE_MESSAGE_DIGEST
-  if(bufSizeFinal > 0) {
-    segment->updateWrittenLength(bufSizeFinal);
+    getSocket()->peekData(buf_, bufSize);
+    streamFilter_->transform(diskAdaptor, segment, buf_, bufSize);
+    bufSize = streamFilter_->getBytesProcessed();
+    getSocket()->readData(buf_, bufSize);
   }
   peerStat_->updateDownloadLength(bufSize);
   getSegmentMan()->updateDownloadSpeedFor(peerStat_);
   bool segmentPartComplete = false;
   // Note that GrowSegment::complete() always returns false.
-  if(transferEncodingDecoder_.isNull() && contentEncodingDecoder_.isNull()) {
+  if(sinkFilterOnly_) {
     if(segment->complete() ||
        segment->getPositionToWrite() == getFileEntry()->getLastOffset()) {
       segmentPartComplete = true;
@@ -203,23 +172,20 @@ bool DownloadCommand::executeInternal() {
     }
   } else {
     off_t loff = getFileEntry()->gtoloff(segment->getPositionToWrite());
-    if(!transferEncodingDecoder_.isNull() &&
-       ((loff == getRequestEndOffset() && transferEncodingDecoder_->finished())
+    if(getFileEntry()->getLength() > 0 && !sinkFilterOnly_ &&
+       ((loff == getRequestEndOffset() && streamFilter_->finished())
         || loff < getRequestEndOffset()) &&
        (segment->complete() ||
         segment->getPositionToWrite() == getFileEntry()->getLastOffset())) {
-      // In this case, transferEncodingDecoder is used and
-      // Content-Length is known.  We check
-      // transferEncodingDecoder_->finished() only if the requested
-      // end offset equals to written position in file local offset;
-      // in other words, data in the requested ranage is all received.
-      // If requested end offset is greater than this segment, then
-      // transferEncodingDecoder_ is not finished in this segment.
+      // In this case, StreamFilter other than *SinkStreamFilter is
+      // used and Content-Length is known.  We check
+      // streamFilter_->finished() only if the requested end offset
+      // equals to written position in file local offset; in other
+      // words, data in the requested ranage is all received.  If
+      // requested end offset is greater than this segment, then
+      // streamFilter_ is not finished in this segment.
       segmentPartComplete = true;
-    } else if((transferEncodingDecoder_.isNull() ||
-               transferEncodingDecoder_->finished()) &&
-              (contentEncodingDecoder_.isNull() ||
-               contentEncodingDecoder_->finished())) {
+    } else if(streamFilter_->finished()) {
       segmentPartComplete = true;
     }
   }
@@ -392,18 +358,18 @@ void DownloadCommand::validatePieceHash(const SharedHandle<Segment>& segment,
   }
 }
 
-#endif // ENABLE_MESSAGE_DIGEST
-
-void DownloadCommand::setTransferEncodingDecoder
-(const SharedHandle<Decoder>& decoder)
+void DownloadCommand::installStreamFilter
+(const SharedHandle<StreamFilter>& streamFilter)
 {
-  this->transferEncodingDecoder_ = decoder;
+  if(streamFilter.isNull()) {
+    return;
+  }
+  streamFilter->installDelegate(streamFilter_);
+  streamFilter_ = streamFilter;
+  sinkFilterOnly_ =
+    util::endsWith(streamFilter_->getName(), SinkStreamFilter::NAME);
 }
 
-void DownloadCommand::setContentEncodingDecoder
-(const SharedHandle<Decoder>& decoder)
-{
-  contentEncodingDecoder_ = decoder;
-}
+#endif // ENABLE_MESSAGE_DIGEST
 
 } // namespace aria2