From efbfe4c006cfb6e0bf3ee4ed57d0657c0cd85c19 Mon Sep 17 00:00:00 2001 From: Tatsuhiro Tsujikawa Date: Mon, 6 Sep 2010 14:29:36 +0000 Subject: 2010-09-06 Tatsuhiro Tsujikawa 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 --- src/ChunkedDecodingStreamFilter.cc | 223 +++++++++++++++++++++++++++++++++++++ 1 file changed, 223 insertions(+) create mode 100644 src/ChunkedDecodingStreamFilter.cc (limited to 'src/ChunkedDecodingStreamFilter.cc') diff --git a/src/ChunkedDecodingStreamFilter.cc b/src/ChunkedDecodingStreamFilter.cc new file mode 100644 index 00000000..87084707 --- /dev/null +++ b/src/ChunkedDecodingStreamFilter.cc @@ -0,0 +1,223 @@ +/* */ +#include "ChunkedDecodingStreamFilter.h" + +#include + +#include "util.h" +#include "message.h" +#include "DlAbortEx.h" +#include "StringFormat.h" +#include "A2STR.h" + +namespace aria2 { + +const std::string ChunkedDecodingStreamFilter::NAME +("ChunkedDecodingStreamFilter"); + +size_t ChunkedDecodingStreamFilter::MAX_BUF_SIZE = 1024*1024; + +ChunkedDecodingStreamFilter::ReadChunkSizeStateHandler* +ChunkedDecodingStreamFilter::readChunkSizeStateHandler_ = + new ChunkedDecodingStreamFilter::ReadChunkSizeStateHandler(); + +ChunkedDecodingStreamFilter::ReadTrailerStateHandler* +ChunkedDecodingStreamFilter::readTrailerStateHandler_ = + new ChunkedDecodingStreamFilter::ReadTrailerStateHandler(); + +ChunkedDecodingStreamFilter::ReadDataStateHandler* +ChunkedDecodingStreamFilter::readDataStateHandler_ = + new ChunkedDecodingStreamFilter::ReadDataStateHandler(); + +ChunkedDecodingStreamFilter::ReadDataEndStateHandler* +ChunkedDecodingStreamFilter::readDataEndStateHandler_ = + new ChunkedDecodingStreamFilter::ReadDataEndStateHandler(); + +ChunkedDecodingStreamFilter::StreamEndStatehandler* +ChunkedDecodingStreamFilter::streamEndStateHandler_ = + new ChunkedDecodingStreamFilter::StreamEndStatehandler(); + +ChunkedDecodingStreamFilter::ChunkedDecodingStreamFilter +(const SharedHandle& delegate): + StreamFilter(delegate), + state_(readChunkSizeStateHandler_), + chunkSize_(0), + bytesProcessed_(0) {} + +ChunkedDecodingStreamFilter::~ChunkedDecodingStreamFilter() {} + +void ChunkedDecodingStreamFilter::init() {} + +bool ChunkedDecodingStreamFilter::readChunkSize +(size_t& inbufOffset, const unsigned char* inbuf, size_t inlen) +{ + size_t pbufSize = buf_.size(); + buf_.append(&inbuf[inbufOffset], &inbuf[inlen]); + std::string::size_type crlfPos = buf_.find(A2STR::CRLF); + if(crlfPos == std::string::npos) { + if(buf_.size() > MAX_BUF_SIZE) { + throw DL_ABORT_EX("Could not find chunk size before buffer got full."); + } + inbufOffset = inlen; + return false; + } + std::string::size_type extPos = buf_.find(A2STR::SEMICOLON_C); + if(extPos == std::string::npos || crlfPos < extPos) { + extPos = crlfPos; + } + chunkSize_ = util::parseULLInt(buf_.substr(0, extPos), 16); + assert(crlfPos+2 > pbufSize); + inbufOffset += crlfPos+2-pbufSize; + buf_.clear(); + if(chunkSize_ == 0) { + state_ = readTrailerStateHandler_; + } else { + state_ = readDataStateHandler_; + } + return true; +} + +bool ChunkedDecodingStreamFilter::readTrailer +(size_t& inbufOffset, const unsigned char* inbuf, size_t inlen) +{ + size_t pbufSize = buf_.size(); + buf_.append(&inbuf[inbufOffset], &inbuf[inlen]); + std::string::size_type crlfcrlfPos = buf_.find("\r\n\r\n"); + if(crlfcrlfPos != std::string::npos) { + // TODO crlfcrlfPos == 0 case? + inbufOffset += crlfcrlfPos+4-pbufSize; + inbufOffset = inlen; + buf_.clear(); + state_ = streamEndStateHandler_; + return true; + } else { + std::string::size_type crlfPos = buf_.find(A2STR::CRLF); + if(crlfPos == std::string::npos) { + if(buf_.size() > MAX_BUF_SIZE) { + throw DL_ABORT_EX + ("Could not find end of stream before buffer got full."); + } + inbufOffset = inlen; + return false; + } else if(crlfPos == 0) { + inbufOffset += crlfPos+2-pbufSize; + buf_.clear(); + state_ = streamEndStateHandler_; + return true; + } else { + if(buf_.size() > MAX_BUF_SIZE) { + throw DL_ABORT_EX + ("Could not find end of stream before buffer got full."); + } + inbufOffset = inlen; + return false; + } + } +} + +bool ChunkedDecodingStreamFilter::readData +(ssize_t& outlen, + size_t& inbufOffset, + const unsigned char* inbuf, + size_t inlen, + const SharedHandle& out, + const SharedHandle& segment) +{ + uint64_t readlen = + std::min(chunkSize_, static_cast(inlen-inbufOffset)); + outlen += getDelegate()->transform(out, segment, inbuf+inbufOffset, readlen); + chunkSize_ -= readlen; + inbufOffset += readlen; + if(chunkSize_ == 0) { + state_ = readDataEndStateHandler_; + return true; + } else { + return false; + } +} + +bool ChunkedDecodingStreamFilter::readDataEnd +(size_t& inbufOffset, const unsigned char* inbuf, size_t inlen) +{ + size_t pbufSize = buf_.size(); + buf_.append(&inbuf[inbufOffset], &inbuf[inlen]); + if(buf_.size() >= 2) { + if(util::startsWith(buf_, A2STR::CRLF)) { + inbufOffset += 2-pbufSize; + buf_.clear(); + state_ = readChunkSizeStateHandler_; + return true; + } else { + throw DL_ABORT_EX("No CRLF at the end of chunk."); + } + } else { + inbufOffset = inlen; + return false; + } +} + +ssize_t ChunkedDecodingStreamFilter::transform +(const SharedHandle& out, + const SharedHandle& segment, + const unsigned char* inbuf, size_t inlen) +{ + size_t inbufOffset = 0; + ssize_t outlen = 0; + while(inbufOffset < inlen) { + ssize_t olen = 0; + bool r = state_->execute + (this, olen, inbufOffset, inbuf, inlen, out, segment); + outlen += olen; + if(!r) { + break; + } + } + bytesProcessed_ = inbufOffset; + return outlen; +} + +bool ChunkedDecodingStreamFilter::finished() +{ + return state_ == streamEndStateHandler_ && getDelegate()->finished(); +} + +void ChunkedDecodingStreamFilter::release() {} + +const std::string& ChunkedDecodingStreamFilter::getName() const +{ + return NAME; +} + +} // namespace aria2 -- cgit