Replace StreamListenSocket with StreamSocket in HttpServer.

net/server/http_server.cc

Index: net/server/http_server.cc
diff --git a/net/server/http_server.cc b/net/server/http_server.cc
index 043e625d6db35eb53ca6f0f1894da648efd5e329..03e0a3fd5ccb016d008bd729f6026882deb62e56 100644
--- a/net/server/http_server.cc
+++ b/net/server/http_server.cc
@@ -17,14 +17,25 @@
 #include "net/server/http_server_request_info.h"
 #include "net/server/http_server_response_info.h"
 #include "net/server/web_socket.h"
-#include "net/socket/tcp_listen_socket.h"
+#include "net/socket/server_socket.h"
+#include "net/socket/stream_socket.h"
+#include "net/socket/tcp_server_socket.h"
 
 namespace net {
 
-HttpServer::HttpServer(const StreamListenSocketFactory& factory,
+HttpServer::HttpServer(scoped_ptr<ServerSocket> server_socket,
                        HttpServer::Delegate* delegate)
-    : delegate_(delegate),
-      server_(factory.CreateAndListen(this)) {
+    : server_socket_(server_socket.Pass()),
+      delegate_(delegate),
+      last_id_(0),
+      weak_ptr_factory_(this) {
+  DCHECK(server_socket_);
+  DoAcceptLoop();
+}
+
+HttpServer::~HttpServer() {
+  STLDeleteContainerPairSecondPointers(
+      id_to_connection_.begin(), id_to_connection_.end());
 }
 
 void HttpServer::AcceptWebSocket(
@@ -33,9 +44,8 @@ void HttpServer::AcceptWebSocket(
   HttpConnection* connection = FindConnection(connection_id);
   if (connection == NULL)
     return;
-
-  DCHECK(connection->web_socket_.get());
-  connection->web_socket_->Accept(request);
+  DCHECK(connection->web_socket());
+  connection->web_socket()->Accept(request);
 }
 
 void HttpServer::SendOverWebSocket(int connection_id,
@@ -43,23 +53,23 @@ void HttpServer::SendOverWebSocket(int connection_id,
   HttpConnection* connection = FindConnection(connection_id);
   if (connection == NULL)
     return;
-  DCHECK(connection->web_socket_.get());
-  connection->web_socket_->Send(data);
+  DCHECK(connection->web_socket());
+  connection->web_socket()->Send(data);
 }
 
 void HttpServer::SendRaw(int connection_id, const std::string& data) {
   HttpConnection* connection = FindConnection(connection_id);
   if (connection == NULL)
     return;
-  connection->Send(data);
+
+  bool writing_in_progress = !connection->write_buf()->IsEmpty();
+  if (connection->write_buf()->Append(data) && !writing_in_progress)
+    DoWriteLoop(connection);
 }
 
 void HttpServer::SendResponse(int connection_id,
                               const HttpServerResponseInfo& response) {
-  HttpConnection* connection = FindConnection(connection_id);
-  if (connection == NULL)
-    return;
-  connection->Send(response);
+  SendRaw(connection_id, response.Serialize());
 }
 
 void HttpServer::Send(int connection_id,
@@ -67,8 +77,9 @@ void HttpServer::Send(int connection_id,
                       const std::string& data,
                       const std::string& content_type) {
   HttpServerResponseInfo response(status_code);
-  response.SetBody(data, content_type);
+  response.SetContentHeaders(data.size(), content_type);
   SendResponse(connection_id, response);
+  SendRaw(connection_id, data);
 }
 
 void HttpServer::Send200(int connection_id,
@@ -86,42 +97,106 @@ void HttpServer::Send500(int connection_id, const std::string& message) {
 }
 
 void HttpServer::Close(int connection_id) {
+  // The call stack might have callbacks which still have the pointer of
+  // connection. Instead of referencing connection with ID all the time,
+  // destroys the connection in next run loop to make sure any pending
+  // callbacks in the call stack return.
+  base::MessageLoopProxy::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&HttpServer::CloseConnection,
+                 weak_ptr_factory_.GetWeakPtr(), connection_id));

[mmenke, 2014/08/15 14:23:16]

This is my remaining concern - this is really weird.  This is a public function,
that the consumer can call directly.

So we can still send it some data for a given connection, after it calls Close()
on it.

I think we really need a reasonable story here.  One option is to check if a
connection is closed in any call that may nest down to this level.  Gets ugly,
and hard to be sure we get them all, but it works.

Or we could delete the connection here, and keep a weak pointer to it and check
that - I prefer that approach.  Open to other ideas.  Suppose we could also make
it an error to close a connection during a callback, but that seems too
draconian.

[byungchul, 2014/08/15 17:12:17]

Took the first approach because delegate_ is private and easy to track all
callbacks.

+}
+
+int HttpServer::GetLocalAddress(IPEndPoint* address) {
+  return server_socket_->GetLocalAddress(address);
+}
+
+void HttpServer::SetReceiveBufferSize(int connection_id, int32 size) {
   HttpConnection* connection = FindConnection(connection_id);
-  if (connection == NULL)
-    return;
+  DCHECK(connection);
+  connection->read_buf()->set_max_buffer_size(size);
+}
 
-  // Initiating close from server-side does not lead to the DidClose call.
-  // Do it manually here.
-  DidClose(connection->socket_.get());
+void HttpServer::SetSendBufferSize(int connection_id, int32 size) {
+  HttpConnection* connection = FindConnection(connection_id);
+  DCHECK(connection);
+  connection->write_buf()->set_max_buffer_size(size);
 }
 
-int HttpServer::GetLocalAddress(IPEndPoint* address) {
-  if (!server_)
-    return ERR_SOCKET_NOT_CONNECTED;
-  return server_->GetLocalAddress(address);
+void HttpServer::DoAcceptLoop() {
+  int rv;
+  do {
+    rv = server_socket_->Accept(&accepted_socket_,
+                                base::Bind(&HttpServer::OnAcceptCompleted,
+                                           weak_ptr_factory_.GetWeakPtr()));
+    if (rv == ERR_IO_PENDING)
+      return;
+    rv = HandleAcceptResult(rv);
+  } while (rv == OK);
 }
 
-void HttpServer::DidAccept(StreamListenSocket* server,
-                           scoped_ptr<StreamListenSocket> socket) {
-  HttpConnection* connection = new HttpConnection(this, socket.Pass());
+void HttpServer::OnAcceptCompleted(int rv) {
+  if (HandleAcceptResult(rv) == OK)
+    DoAcceptLoop();
+}
+
+int HttpServer::HandleAcceptResult(int rv) {
+  if (rv < 0) {
+    LOG(ERROR) << "Accept error: rv=" << rv;
+    return rv;
+  }
+
+  HttpConnection* connection =
+      new HttpConnection(++last_id_, accepted_socket_.Pass());
   id_to_connection_[connection->id()] = connection;
-  // TODO(szym): Fix socket access. Make HttpConnection the Delegate.
-  socket_to_connection_[connection->socket_.get()] = connection;
+  DoReadLoop(connection);
+  return OK;
 }
 
-void HttpServer::DidRead(StreamListenSocket* socket,
-                         const char* data,
-                         int len) {
-  HttpConnection* connection = FindConnection(socket);
-  DCHECK(connection != NULL);
-  if (connection == NULL)
+void HttpServer::DoReadLoop(HttpConnection* connection) {
+  int rv;
+  do {
+    HttpConnection::ReadIOBuffer* read_buf = connection->read_buf();
+    // Increases read buffer size if necessary.
+    if (read_buf->RemainingCapacity() == 0 && !read_buf->IncreaseCapacity()) {
+      Close(connection->id());
+      return;
+    }
+
+    rv = connection->socket()->Read(
+        read_buf,
+        read_buf->RemainingCapacity(),
+        base::Bind(&HttpServer::OnReadCompleted,
+                   weak_ptr_factory_.GetWeakPtr(), connection->id()));
+    if (rv == ERR_IO_PENDING)
+      return;
+    rv = HandleReadResult(connection, rv);
+  } while (rv == OK);
+}
+
+void HttpServer::OnReadCompleted(int connection_id, int rv) {
+  HttpConnection* connection = FindConnection(connection_id);
+  if (!connection)  // It might be closed right before by write error.
     return;
 
-  connection->recv_data_.append(data, len);
-  while (connection->recv_data_.length()) {
-    if (connection->web_socket_.get()) {
+  if (HandleReadResult(connection, rv) == OK)
+    DoReadLoop(connection);
+}
+
+int HttpServer::HandleReadResult(HttpConnection* connection, int rv) {
+  if (rv <= 0) {
+    Close(connection->id());
+    return rv == 0 ? ERR_CONNECTION_CLOSED : rv;
+  }
+
+  HttpConnection::ReadIOBuffer* read_buf = connection->read_buf();
+  read_buf->DidRead(rv);
+
+  // Handles http requests or websocket messages.
+  while (read_buf->GetSize() > 0) {
+    if (connection->web_socket()) {
       std::string message;
-      WebSocket::ParseResult result = connection->web_socket_->Read(&message);
+      WebSocket::ParseResult result = connection->web_socket()->Read(&message);
       if (result == WebSocket::FRAME_INCOMPLETE)
         break;
 
@@ -136,62 +211,99 @@ void HttpServer::DidRead(StreamListenSocket* socket,
 
     HttpServerRequestInfo request;
     size_t pos = 0;
-    if (!ParseHeaders(connection, &request, &pos))
+    if (!ParseHeaders(read_buf->StartOfBuffer(), read_buf->GetSize(),
+                      &request, &pos)) {
       break;
+    }
 
     // Sets peer address if exists.
-    socket->GetPeerAddress(&request.peer);
+    connection->socket()->GetPeerAddress(&request.peer);
 
     if (request.HasHeaderValue("connection", "upgrade")) {
-      connection->web_socket_.reset(WebSocket::CreateWebSocket(connection,
-                                                               request,
-                                                               &pos));
-
-      if (!connection->web_socket_.get())  // Not enough data was received.
+      scoped_ptr<WebSocket> websocket(
+          WebSocket::CreateWebSocket(this, connection, request, &pos));
+      if (!websocket)  // Not enough data was received.
         break;
+      connection->SetWebSocket(websocket.Pass());
+      read_buf->DidConsume(pos);
       delegate_->OnWebSocketRequest(connection->id(), request);
-      connection->Shift(pos);
       continue;
     }
 
     const char kContentLength[] = "content-length";
-    if (request.headers.count(kContentLength)) {
+    if (request.headers.count(kContentLength) > 0) {
       size_t content_length = 0;
       const size_t kMaxBodySize = 100 << 20;
       if (!base::StringToSizeT(request.GetHeaderValue(kContentLength),
                                &content_length) ||
           content_length > kMaxBodySize) {
-        connection->Send(HttpServerResponseInfo::CreateFor500(
-            "request content-length too big or unknown: " +
-            request.GetHeaderValue(kContentLength)));
-        DidClose(socket);
+        SendResponse(connection->id(),
+                     HttpServerResponseInfo::CreateFor500(
+                         "request content-length too big or unknown: " +
+                         request.GetHeaderValue(kContentLength)));
+        Close(connection->id());
         break;
       }
 
-      if (connection->recv_data_.length() - pos < content_length)
+      if (read_buf->GetSize() - pos < content_length)
         break;  // Not enough data was received yet.
-      request.data = connection->recv_data_.substr(pos, content_length);
+      request.data.assign(read_buf->StartOfBuffer() + pos, content_length);
       pos += content_length;
     }
 
+    read_buf->DidConsume(pos);
     delegate_->OnHttpRequest(connection->id(), request);
-    connection->Shift(pos);
   }
+
+  return OK;
 }
 
-void HttpServer::DidClose(StreamListenSocket* socket) {
-  HttpConnection* connection = FindConnection(socket);
-  DCHECK(connection != NULL);
-  id_to_connection_.erase(connection->id());
-  socket_to_connection_.erase(connection->socket_.get());
-  delete connection;
+void HttpServer::DoWriteLoop(HttpConnection* connection) {
+  int rv = OK;
+  HttpConnection::QueuedWriteIOBuffer* write_buf = connection->write_buf();
+  while (rv == OK && write_buf->GetSizeToWrite() > 0) {
+    rv = connection->socket()->Write(
+        write_buf,
+        write_buf->GetSizeToWrite(),
+        base::Bind(&HttpServer::OnWriteCompleted,
+                   weak_ptr_factory_.GetWeakPtr(), connection->id()));
+    if (rv == ERR_IO_PENDING || rv == OK)
+      return;
+    rv = HandleWriteResult(connection, rv);
+  }
 }
 
-HttpServer::~HttpServer() {
-  STLDeleteContainerPairSecondPointers(
-      id_to_connection_.begin(), id_to_connection_.end());
+void HttpServer::OnWriteCompleted(int connection_id, int rv) {
+  HttpConnection* connection = FindConnection(connection_id);
+  if (!connection)  // It might be closed right before by read error.
+    return;
+
+  if (HandleWriteResult(connection, rv) == OK)
+    DoWriteLoop(connection);
+}
+
+int HttpServer::HandleWriteResult(HttpConnection* connection, int rv) {
+  if (rv < 0) {
+    Close(connection->id());
+    return rv;
+  }
+
+  connection->write_buf()->DidConsume(rv);
+  return OK;
+}
+
+void HttpServer::CloseConnection(int connection_id) {
+  HttpConnection* connection = FindConnection(connection_id);
+  if (connection == NULL)
+    return;
+
+  id_to_connection_.erase(connection_id);
+  delete connection;
+  delegate_->OnClose(connection_id);
 }
 
+namespace {
+
 //
 // HTTP Request Parser
 // This HTTP request parser uses a simple state machine to quickly parse
@@ -255,17 +367,19 @@ int charToInput(char ch) {
   return INPUT_DEFAULT;
 }
 
-bool HttpServer::ParseHeaders(HttpConnection* connection,
+}  // namespace
+
+bool HttpServer::ParseHeaders(const char* data,
+                              size_t data_len,
                               HttpServerRequestInfo* info,
                               size_t* ppos) {
   size_t& pos = *ppos;
-  size_t data_len = connection->recv_data_.length();
   int state = ST_METHOD;
   std::string buffer;
   std::string header_name;
   std::string header_value;
   while (pos < data_len) {
-    char ch = connection->recv_data_[pos++];
+    char ch = data[pos++];
     int input = charToInput(ch);
     int next_state = parser_state[state][input];
 
@@ -337,11 +451,4 @@ HttpConnection* HttpServer::FindConnection(int connection_id) {
   return it->second;
 }
 
-HttpConnection* HttpServer::FindConnection(StreamListenSocket* socket) {
-  SocketToConnectionMap::iterator it = socket_to_connection_.find(socket);
-  if (it == socket_to_connection_.end())
-    return NULL;
-  return it->second;
-}
-
 }  // namespace net