Replace StreamListenSocket with StreamSocket in HttpServer.

net/server/http_connection.cc

Index: net/server/http_connection.cc
diff --git a/net/server/http_connection.cc b/net/server/http_connection.cc
index d433012cd651c7b7462972c3835cb818a9a78d8d..b53a6f6d6407bec07de5925930486c9c3faeeedf 100644
--- a/net/server/http_connection.cc
+++ b/net/server/http_connection.cc
@@ -4,44 +4,161 @@
 
 #include "net/server/http_connection.h"
 
-#include "net/server/http_server.h"
-#include "net/server/http_server_response_info.h"
 #include "net/server/web_socket.h"
-#include "net/socket/stream_listen_socket.h"
+#include "net/socket/stream_socket.h"
 
 namespace net {
 
-int HttpConnection::last_id_ = 0;
+namespace {
 
-void HttpConnection::Send(const std::string& data) {
-  if (!socket_.get())
+const int kInitialReadBufSize = 1024;
+const int kMinimumReadBufSize = 128;
+const int kCapacityIncreaseFactor = 2;
+
+const int kDefaultPendingDataLimit = 1 * 1024 * 1024;  // 1 Mbytes.
+
+}  // namespace
+
+HttpConnection::ReadIOBuffer::ReadIOBuffer()
+    : base_(new GrowableIOBuffer()),
+      capacity_limit_(kDefaultPendingDataLimit) {
+  SetCapacity(kInitialReadBufSize);
+}
+
+HttpConnection::ReadIOBuffer::~ReadIOBuffer() {
+  data_ = NULL;  // base_ owns data_.
+}
+
+IOBuffer* HttpConnection::ReadIOBuffer::GetUnusedIOBuffer() const {
+  return base_.get();
+}
+
+int HttpConnection::ReadIOBuffer::GetUnusedCapacity() const {
+  return base_->RemainingCapacity();
+}
+
+int HttpConnection::ReadIOBuffer::GetCapacity() const {
+  return base_->capacity();
+}
+
+void HttpConnection::ReadIOBuffer::SetCapacity(int capacity) {
+  base_->SetCapacity(capacity);
+  data_ = base_->StartOfBuffer();

[mmenke, 2014/06/06 15:41:03]

This looks wrong to me, when there's unconsumed data.  Should either copy
unconsumed data to the start of the bugger (And update offset_ accordingly)
first, or set data_ to be base_->StartOfBuffer() + [old value of data_ -
base_->StartOfBuffer].

This code is sufficiently complicated that I don't think we should land it prior
to adding unit tests.

[byungchul, 2014/06/06 19:17:35]

Fixed and added unittests.

+}
+
+bool HttpConnection::ReadIOBuffer::IncreaseCapacity() {
+  if (GetCapacity() > capacity_limit_) {
+    LOG(ERROR) << "Too large read data is pending: capacity=" << GetCapacity()
+               << ", capacity_limit=" << capacity_limit_
+               << ", consumed=" << base_->offset();
+    return false;
+  }
+  SetCapacity(GetCapacity() * kCapacityIncreaseFactor);
+  return true;
+}
+
+int HttpConnection::ReadIOBuffer::GetUnconsumedSize() const {
+  DCHECK_GE(base_->data(), data_);
+  return base_->data() - data_;
+}
+
+void HttpConnection::ReadIOBuffer::DidRead(int bytes) {
+  DCHECK_LE(bytes, GetUnusedCapacity());
+  base_->set_offset(base_->offset() + bytes);
+}
+
+void HttpConnection::ReadIOBuffer::DidConsume(int bytes) {
+  DCHECK_LE(bytes, GetUnconsumedSize());
+  if (bytes < GetUnconsumedSize()) {
+    data_ += bytes;
     return;
-  socket_->Send(data);
+  }
+  // Since bytes == GetUnconsumedSize() here, base_->offset() is the size of
+  // consumed data, and all data has been consumed.  No need to keep consumed
+  // data any more because no data will be moved.
+  // If capacity is too big, reduce it.
+  if (GetCapacity() > kMinimumReadBufSize
+      && GetCapacity() > base_->offset() * kCapacityIncreaseFactor) {
+    SetCapacity(GetCapacity() / kCapacityIncreaseFactor);
+  }
+  base_->set_offset(0);
+}
+
+HttpConnection::PendingWriteIOBuffer::PendingWriteIOBuffer()
+    : total_size_(0),
+      total_size_limit_(kDefaultPendingDataLimit) {
+}
+
+HttpConnection::PendingWriteIOBuffer::~PendingWriteIOBuffer() {
 }
 
-void HttpConnection::Send(const char* bytes, int len) {
-  if (!socket_.get())
+bool HttpConnection::PendingWriteIOBuffer::IsEmpty() const {
+  return pending_data_.empty();
+}
+
+bool HttpConnection::PendingWriteIOBuffer::Append(const std::string& data) {
+  if (data.empty()) {
+    return true;
+  }
+
+  if (total_size_ + static_cast<int>(data.size()) > total_size_limit_) {
+    LOG(ERROR) << "Too large write data is pending: size="
+               << total_size_ + data.size()
+               << ", size_limit=" << total_size_limit_;
+    return false;
+  }
+
+  pending_data_.push(data);
+  total_size_ += data.size();
+
+  // If new data is the first pending data, updates data_.
+  if (pending_data_.size() == 1) {
+    data_ = const_cast<char*>(pending_data_.front().data());
+  }
+  return true;
+}
+
+void HttpConnection::PendingWriteIOBuffer::DidConsume(int size) {
+  DCHECK_LE(size, total_size_);
+  DCHECK_LE(size, GetSizeToWrite());
+  if (size == 0) {
     return;
-  socket_->Send(bytes, len);
+  }
+
+  if (size < GetSizeToWrite()) {
+    data_ += size;
+  } else {  // size == GetSizeToWrite(). Updates data_ to next pending data.
+    pending_data_.pop();
+    data_ = IsEmpty() ? NULL : const_cast<char*>(pending_data_.front().data());
+  }
+  total_size_ -= size;
 }
 
-void HttpConnection::Send(const HttpServerResponseInfo& response) {
-  Send(response.Serialize());
+int HttpConnection::PendingWriteIOBuffer::GetSizeToWrite() const {
+  if (IsEmpty()) {
+    DCHECK_EQ(total_size_, 0);
+    return 0;
+  }
+  DCHECK_GE(data_, pending_data_.front().data());
+  int consumed = static_cast<int>(data_ - pending_data_.front().data());
+  DCHECK_GT(static_cast<int>(pending_data_.front().size()), consumed);
+  return pending_data_.front().size() - consumed;
 }
 
-HttpConnection::HttpConnection(HttpServer* server,
-                               scoped_ptr<StreamListenSocket> sock)
-    : server_(server),
-      socket_(sock.Pass()) {
-  id_ = last_id_++;
+HttpConnection::HttpConnection(int id,
+                               scoped_ptr<StreamSocket> socket)
+    : id_(id),
+      socket_(socket.Pass()),
+      read_buf_(new ReadIOBuffer()),
+      write_buf_(new PendingWriteIOBuffer()) {
 }
 
 HttpConnection::~HttpConnection() {
-  server_->delegate_->OnClose(id_);
 }
 
-void HttpConnection::Shift(int num_bytes) {
-  recv_data_ = recv_data_.substr(num_bytes);
+void HttpConnection::SetWebSocket(scoped_ptr<WebSocket> web_socket) {
+  DCHECK(!web_socket_);
+  web_socket_ = web_socket.Pass();
 }
 
 }  // namespace net