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..0807326e2bada919db89e0842edd75ed2e8e143c 100644
--- a/net/server/http_connection.cc
+++ b/net/server/http_connection.cc
@@ -4,44 +4,169 @@
 
 #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;
+HttpConnection::ReadIOBuffer::ReadIOBuffer()
+    : base_(new GrowableIOBuffer()),
+      capacity_limit_(kDefaultCapacityLimit) {
+  base_->SetCapacity(kInitialBufSize);
+  data_ = base_->StartOfBuffer();
+}
+
+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) {
+  DCHECK(data_);
+  int consumed_size = static_cast<int>(data_ - base_->StartOfBuffer());
+  base_->SetCapacity(capacity);
+  data_ = base_->StartOfBuffer() + consumed_size;
+}
+
+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;
+  }
+
+  int new_capacity = GetCapacity() * kCapacityIncreaseFactor;
+  if (new_capacity > capacity_limit_) {
+    new_capacity = capacity_limit_;
+  }
+  SetCapacity(new_capacity);
+  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::Send(const std::string& data) {
-  if (!socket_.get())
+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() > kMinimumBufSize
+      && GetCapacity() > base_->offset() * kCapacityIncreaseFactor) {
+    int new_capacity = GetCapacity() / kCapacityIncreaseFactor;
+    if (new_capacity < kMinimumBufSize) {
+      new_capacity = kMinimumBufSize;
+    }
+    // realloc() within GrowableIOBuffer::SetCapacity() could move data even
+    // when size is reduced. Free internal data first to guarantee no data move.
+    base_->SetCapacity(0);
+    base_->SetCapacity(new_capacity);
+  } else {
+    base_->set_offset(0);
+  }
+  data_ = base_->StartOfBuffer();
+}
+
+HttpConnection::PendingWriteIOBuffer::PendingWriteIOBuffer()
+    : total_size_(0),
+      total_size_limit_(kDefaultTotalSizeLimit) {
+}
+
+HttpConnection::PendingWriteIOBuffer::~PendingWriteIOBuffer() {
+  data_ = NULL;  // pending_data_ owns data_.
 }
 
-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