Refine the streaming, add UT, add error handling.

webcc/body.h

@@ -169,6 +169,10 @@ public:
 
   void Dump(std::ostream& os, const std::string& prefix) const override;
 
+  const Path& path() const {
+    return path_;
+  }
+
   // Move (or rename) the file.
   // Used to move the streamed file of the received message to a new place.
   // Applicable to both client and server.

webcc/client.cc

@@ -16,15 +16,19 @@ Client::Client()
 }
 
 Error Client::Request(RequestPtr request, bool connect, bool stream) {
-  io_context_.restart();
-
-  response_.reset(new Response{});
-  response_parser_.Init(response_.get(), stream);
-
   closed_ = false;
   timer_canceled_ = false;
   error_ = Error{};
 
+  response_.reset(new Response{});
+
+  if (!response_parser_.Init(response_.get(), stream)) {
+    // Failed to generate the temp file for streaming.
+    // I don't know when this would happen. Keep the error handling here just
+    // for preciseness.
+    return Error{ Error::kFileError, "Streaming temp file error" };
+  }
+
   if (buffer_.size() != buffer_size_) {
     LOG_VERB("Resize buffer: %u -> %u.", buffer_.size(), buffer_size_);
     buffer_.resize(buffer_size_);
@@ -44,6 +48,8 @@ Error Client::Request(RequestPtr request, bool connect, bool stream) {
     response_parser_.set_ignroe_body(false);
   }
 
+  io_context_.restart();
+
   if (connect) {
     // No existing socket connection was specified, create a new one.
     Connect(request);

webcc/client.h

@@ -54,9 +54,21 @@ public:
   // Close the socket.
   void Close();
 
-  ResponsePtr response() const { return response_; }
+  ResponsePtr response() const {
+    return response_;
+  }
 
-  bool closed() const { return closed_; }
+  // Reset response object.
+  // Used to make sure the response object will released even the client object
+  // itself will be cached for keep-alive purpose.
+  void Reset() {
+    response_.reset();
+    response_parser_.Init(nullptr, false);
+  }
+
+  bool closed() const {
+    return closed_;
+  }
 
 private:
   void Connect(RequestPtr request);

webcc/client_session.cc

@@ -232,7 +232,11 @@ ResponsePtr ClientSession::Send(RequestPtr request, bool stream) {
     }
   }
 
-  return client->response();
+  auto response = client->response();
+  // The client object might be cached in the pool.
+  // Reset to make sure it won't keep a reference to the response object.
+  client->Reset();
+  return response;
 }
 
 }  // namespace webcc

webcc/message.cc

@@ -39,11 +39,7 @@ const std::string& Message::data() const {
 }
 
 std::shared_ptr<FileBody> Message::file_body() const {
-  auto file_body = std::dynamic_pointer_cast<FileBody>(body_);
-  if (!file_body) {
-    throw Error{ Error::kDataError, "Not a file body" };
-  }
-  return file_body;
+  return std::dynamic_pointer_cast<FileBody>(body_);
 }
 
 bool Message::IsConnectionKeepAlive() const {

webcc/message.h

@@ -27,11 +27,11 @@ public:
   }
 
   // Get the data from the (string) body.
-  // Empty string will be returned if the body is not a StringBody.
+  // Return empty string if the body is not a StringBody.
   const std::string& data() const;
 
   // Get the body as a FileBody.
-  // Exception Error::kDataError will be thrown if the body is not a FileBody.
+  // Return null if the body is not a FileBody.
   std::shared_ptr<FileBody> file_body() const;
 
   // ---------------------------------------------------------------------------

webcc/parser.cc

@@ -17,65 +17,48 @@ namespace webcc {
 
 // -----------------------------------------------------------------------------
 
-ParseHandler::ParseHandler(Message* message, bool stream)
-    : message_(message), content_length_(kInvalidLength), stream_(stream),
-      streamed_size_(0) {
-  if (stream_) {
-    try {
-      temp_path_ = bfs::temp_directory_path() / bfs::unique_path();
-    } catch (const bfs::filesystem_error&) {
-      throw Error{ Error::kFileError, "Cannot generate temp file path" };
-    }
-
-    ofstream_.open(temp_path_, std::ios::binary);
-
-    if (ofstream_.fail()) {
-      throw Error{ Error::kFileError, "Cannot open the temp file" };
-    }
-  }
+ParseHandlerBase::ParseHandlerBase(Message* message)
+    : message_(message), content_length_(kInvalidLength) {
 }
 
-ParseHandler::~ParseHandler() {
-}
-
-void ParseHandler::OnStartLine(const std::string& start_line) {
+void ParseHandlerBase::OnStartLine(const std::string& start_line) {
   message_->set_start_line(start_line);
 }
 
-void ParseHandler::OnContentLength(std::size_t content_length) {
+void ParseHandlerBase::OnContentLength(std::size_t content_length) {
   content_length_ = content_length;
-
-  if (!stream_) {
-    // Reserve memory to avoid frequent reallocation when append.
-    // TODO: Don't know if it's really necessary.
-    try {
-      content_.reserve(content_length_);
-    } catch (const std::exception& e) {
-      LOG_ERRO("Failed to reserve content memory: %s.", e.what());
-    }
-  }
 }
 
-void ParseHandler::OnHeader(Header&& header) {
+void ParseHandlerBase::OnHeader(Header&& header) {
   message_->SetHeader(std::move(header));
 }
 
-void ParseHandler::AddContent(const char* data, std::size_t count) {
-  if (stream_) {
-    ofstream_.write(data, count);
-    streamed_size_ += count;
-  } else {
-    content_.append(data, count);
+bool ParseHandlerBase::IsCompressed() const {
+  return message_->GetContentEncoding() != ContentEncoding::kUnknown;
+}
+
+// -----------------------------------------------------------------------------
+
+ParseHandler::ParseHandler(Message* message) : ParseHandlerBase(message) {
+}
+
+void ParseHandler::OnContentLength(std::size_t content_length) {
+  ParseHandlerBase::OnContentLength(content_length);
+
+  // Reserve memory to avoid frequent reallocation when append.
+  try {
+    content_.reserve(content_length_);
+  } catch (const std::exception& e) {
+    LOG_ERRO("Failed to reserve content memory: %s.", e.what());
   }
 }
 
+void ParseHandler::AddContent(const char* data, std::size_t count) {
+  content_.append(data, count);
+}
+
 void ParseHandler::AddContent(const std::string& data) {
-  if (stream_) {
-    ofstream_ << data;
-    streamed_size_ += data.size();
-  } else {
-    content_.append(data);
-  }
+  content_.append(data);
 }
 
 bool ParseHandler::IsFixedContentFull() const {
@@ -85,11 +68,7 @@ bool ParseHandler::IsFixedContentFull() const {
     return false;
   }
 
-  if (stream_) {
-    return streamed_size_ >= content_length_;
-  } else {
-    return content_.length() >= content_length_;
-  }
+  return content_.length() >= content_length_;
 }
 
 bool ParseHandler::Finish() {
@@ -97,43 +76,88 @@ bool ParseHandler::Finish() {
   // Could be `0` (empty body and `Content-Length : 0`).
   message_->set_content_length(content_length_);
 
-  if (!stream_ && content_.empty()) {
+  if (content_.empty()) {
     // The call to message_->SetBody() is not necessary since message is
     // always initialized with an empty body.
     return true;
   }
 
-  BodyPtr body;
-
-  if (stream_) {
-    ofstream_.close();
-
-    // Create a file body based on the streamed temp file.
-    body = std::make_shared<FileBody>(temp_path_, true);
-
-    // TODO: Compress
-
-  } else {
-    body = std::make_shared<StringBody>(std::move(content_), IsCompressed());
+  auto body = std::make_shared<StringBody>(std::move(content_), IsCompressed());
 
 #if WEBCC_ENABLE_GZIP
-    LOG_INFO("Decompress the HTTP content...");
-    if (!body->Decompress()) {
-      LOG_ERRO("Cannot decompress the HTTP content!");
-      return false;
-    }
-#else
-    LOG_WARN("Compressed HTTP content remains untouched.");
-#endif  // WEBCC_ENABLE_GZIP
+  LOG_INFO("Decompress the HTTP content...");
+  if (!body->Decompress()) {
+    LOG_ERRO("Cannot decompress the HTTP content!");
+    return false;
   }
+#else
+  LOG_WARN("Compressed HTTP content remains untouched.");
+#endif  // WEBCC_ENABLE_GZIP
 
   message_->SetBody(body, false);
+  return true;
+}
+
+// -----------------------------------------------------------------------------
+
+StreamedParseHandler::StreamedParseHandler(Message* message)
+    : ParseHandlerBase(message) {
+}
+
+bool StreamedParseHandler::Init() {
+  try {
+    temp_path_ = bfs::temp_directory_path() / bfs::unique_path();
+    LOG_VERB("Generate a temp path for streaming: %s",
+             temp_path_.string().c_str());
+  } catch (const bfs::filesystem_error&) {
+    LOG_ERRO("Failed to generate temp path: %s", temp_path_.string().c_str());
+    return false;
+  }
+
+  ofstream_.open(temp_path_, std::ios::binary);
+
+  if (ofstream_.fail()) {
+    LOG_ERRO("Failed to open the temp file: %s", temp_path_.string().c_str());
+    return false;
+  }
 
   return true;
 }
 
-bool ParseHandler::IsCompressed() const {
-  return message_->GetContentEncoding() != ContentEncoding::kUnknown;
+void StreamedParseHandler::AddContent(const char* data, std::size_t count) {
+  ofstream_.write(data, count);
+  streamed_size_ += count;
+}
+
+void StreamedParseHandler::AddContent(const std::string& data) {
+  ofstream_ << data;
+  streamed_size_ += data.size();
+}
+
+bool StreamedParseHandler::IsFixedContentFull() const {
+  if (content_length_ == kInvalidLength) {
+    // Shouldn't be here.
+    // See Parser::ParseFixedContent().
+    return false;
+  }
+
+  return streamed_size_ >= content_length_;
+}
+
+bool StreamedParseHandler::Finish() {
+  // Could be `kInvalidLength` (chunked).
+  // Could be `0` (empty body and `Content-Length : 0`).
+  message_->set_content_length(content_length_);
+
+  ofstream_.close();
+
+  // Create a file body based on the streamed temp file.
+  auto body = std::make_shared<FileBody>(temp_path_, true);
+
+  // TODO: Compress
+
+  message_->SetBody(body, false);
+  return true;
 }
 
 // -----------------------------------------------------------------------------
@@ -147,9 +171,21 @@ Parser::Parser()
       finished_(false) {
 }
 
-void Parser::Init(Message* message, bool stream) {
+bool Parser::Init(Message* message, bool stream) {
   Reset();
-  handler_.reset(new ParseHandler{ message, stream });
+
+  if (stream) {
+    handler_.reset(new StreamedParseHandler{ message });
+  } else {
+    handler_.reset(new ParseHandler{ message });
+  }
+
+  if (!handler_->Init()) {
+    // Failed to generate temp file for streaming.
+    return false;
+  }
+
+  return true;
 }
 
 bool Parser::Parse(const char* data, std::size_t length) {

webcc/parser.h

@@ -14,13 +14,13 @@ class Message;
 
 // -----------------------------------------------------------------------------
 
-class ParseHandler {
+class ParseHandlerBase {
 public:
-  // If |stream| is true, the data will be streamed to a temp file, and the
-  // body of the message will be FileBody instead of StringBody.
-  ParseHandler(Message* message, bool stream = false);
+  ParseHandlerBase(Message* message);
 
-  ~ParseHandler();
+  virtual ~ParseHandlerBase() = default;
+
+  virtual bool Init() = 0;
 
   std::size_t content_length() const {
     return content_length_;
@@ -28,29 +28,67 @@ public:
 
   void OnStartLine(const std::string& start_line);
 
-  void OnContentLength(std::size_t content_length);
-
   void OnHeader(Header&& header);
 
-  void AddContent(const char* data, std::size_t count);
+  virtual void OnContentLength(std::size_t content_length);
 
-  void AddContent(const std::string& data);
+  virtual void AddContent(const char* data, std::size_t count) = 0;
 
-  bool IsFixedContentFull() const;
+  virtual void AddContent(const std::string& data) = 0;
 
-  bool Finish();
+  virtual bool IsFixedContentFull() const = 0;
 
-private:
+  virtual bool Finish() = 0;
+
+protected:
   bool IsCompressed() const;
 
-private:
+protected:
   Message* message_;
-
   std::size_t content_length_;
-  std::string content_;
+};
 
-  bool stream_;
-  std::size_t streamed_size_;
+class ParseHandler : public ParseHandlerBase {
+public:
+  explicit ParseHandler(Message* message);
+
+  ~ParseHandler() override = default;
+
+  bool Init() override {
+    return true;
+  }
+
+  void OnContentLength(std::size_t content_length) override;
+
+  void AddContent(const char* data, std::size_t count) override;
+  void AddContent(const std::string& data) override;
+
+  bool IsFixedContentFull() const override;
+  bool Finish() override;
+
+private:
+  std::string content_;
+};
+
+// If |stream| is true, the data will be streamed to a temp file, and the
+// body of the message will be FileBody instead of StringBody.
+class StreamedParseHandler : public ParseHandlerBase {
+public:
+  explicit StreamedParseHandler(Message* message);
+
+  ~StreamedParseHandler() override = default;
+
+  // Generate a temp file.
+  bool Init() override;
+
+  void AddContent(const char* data, std::size_t count) override;
+  void AddContent(const std::string& data) override;
+
+  bool IsFixedContentFull() const override;
+  bool Finish() override;
+
+private:
+  std::size_t streamed_size_ = 0;
   boost::filesystem::ofstream ofstream_;
   Path temp_path_;
 };
@@ -66,7 +104,7 @@ public:
   Parser(const Parser&) = delete;
   Parser& operator=(const Parser&) = delete;
 
-  void Init(Message* message, bool stream = false);
+  bool Init(Message* message, bool stream = false);
 
   bool finished() const {
     return finished_;
@@ -103,7 +141,7 @@ protected:
   bool Finish();
 
 protected:
-  std::unique_ptr<ParseHandler> handler_;
+  std::unique_ptr<ParseHandlerBase> handler_;
 
   // Data waiting to be parsed.
   std::string pending_data_;

webcc/request_parser.cc

@@ -13,9 +13,13 @@ namespace webcc {
 RequestParser::RequestParser() : request_(nullptr) {
 }
 
-void RequestParser::Init(Request* request) {
-  Parser::Init(request);
+bool RequestParser::Init(Request* request, bool stream) {
+  if (!Parser::Init(request, stream)) {
+    return false;
+  }
+
   request_ = request;
+  return true;
 }
 
 bool RequestParser::ParseStartLine(const std::string& line) {

webcc/request_parser.h

@@ -15,7 +15,7 @@ public:
 
   ~RequestParser() override = default;
 
-  void Init(Request* request);
+  bool Init(Request* request, bool stream = false);
 
 private:
   bool ParseStartLine(const std::string& line) override;

webcc/response_parser.cc

@@ -41,9 +41,13 @@ void SplitStartLine(const std::string& line, std::vector<std::string>* parts) {
 ResponseParser::ResponseParser() : response_(nullptr) {
 }
 
-void ResponseParser::Init(Response* response, bool stream) {
-  Parser::Init(response, stream);
+bool ResponseParser::Init(Response* response, bool stream) {
+  if (!Parser::Init(response, stream)) {
+    return false;
+  }
+
   response_ = response;
+  return true;
 }
 
 bool ResponseParser::ParseStartLine(const std::string& line) {

webcc/response_parser.h

@@ -15,7 +15,7 @@ public:
 
   ~ResponseParser() override = default;
 
-  void Init(Response* response, bool stream = false);
+  bool Init(Response* response, bool stream = false);
 
   void set_ignroe_body(bool ignroe_body) {
     ignroe_body_ = ignroe_body;