Replace StreamListenSocket with StreamSocket in HttpServer.

net/server/http_server.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/server/http_server.h"
 
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/stl_util.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/sys_byteorder.h"
 #include "build/build_config.h"
 #include "net/base/net_errors.h"
 #include "net/server/http_connection.h"
 #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,
+namespace {
+
+// A helper function to delete connection in next run loop.
+void DeleteConnection(HttpConnection* connection) {
+  delete connection;
+}
+
+}  // namespace
+
+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());
+}
+
+base::WeakPtr<HttpServer> HttpServer::GetWeakPtr() {
+  return weak_ptr_factory_.GetWeakPtr();
 }
 
 void HttpServer::AcceptWebSocket(
     int connection_id,
     const HttpServerRequestInfo& request) {
   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,
                                    const std::string& data) {
   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)

[mmenke, 2014/06/04 16:29:50]

This seems like a weird division of labor to me.  Seems much more natural to put
all this logic related to reading/sending (And drive the WriteLoop) in the
HttpConnection itself, rather than in the HttpServer class.

Same with DidRead.

That way, all logic involved in how to handle appending data is in one place,
instead of strewn across both files, and having this class muck about the guts
of the HttpConnection.

Alternatively, HttpConnection could just be a struct with no internal logic,
rather than a class.

[byungchul, 2014/06/04 17:41:28]

Yes, current HttpConnection is just a struct though it is still named "class".
No internal logic for HttpConnection itself except ctor.

+    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,
                       HttpStatusCode status_code,
                       const std::string& data,
                       const std::string& content_type) {
   HttpServerResponseInfo response(status_code);
   response.SetBody(data, content_type);
   SendResponse(connection_id, response);
 }
@@ skipping 10 matching lines @@
 
 void HttpServer::Send500(int connection_id, const std::string& message) {
   SendResponse(connection_id, HttpServerResponseInfo::CreateFor500(message));
 }
 
 void HttpServer::Close(int connection_id) {
   HttpConnection* connection = FindConnection(connection_id);
   if (connection == NULL)
     return;
 
-  // Initiating close from server-side does not lead to the DidClose call.
-  // Do it manually here.
-  DidClose(connection->socket_.get());
+  id_to_connection_.erase(connection->id());
+  delegate_->OnClose(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(&DeleteConnection, base::Unretained(connection)));
 }
 
 int HttpServer::GetLocalAddress(IPEndPoint* address) {
-  if (!server_)
-    return ERR_SOCKET_NOT_CONNECTED;
-  return server_->GetLocalAddress(address);
+  return server_socket_->GetLocalAddress(address);
 }
 
-void HttpServer::DidAccept(StreamListenSocket* server,
-                           scoped_ptr<StreamListenSocket> socket) {
-  HttpConnection* connection = new HttpConnection(this, socket.Pass());
-  id_to_connection_[connection->id()] = connection;
-  // TODO(szym): Fix socket access. Make HttpConnection the Delegate.
-  socket_to_connection_[connection->socket_.get()] = connection;
+void HttpServer::SetReceiveBufferSize(int connection_id, int32 size) {
+  HttpConnection* connection = FindConnection(connection_id);
+  if (connection == NULL)
+    return;
+  connection->read_buf()->set_capacity_limit(size);
 }
 
-void HttpServer::DidRead(StreamListenSocket* socket,
-                         const char* data,
-                         int len) {
-  HttpConnection* connection = FindConnection(socket);
-  DCHECK(connection != NULL);
+void HttpServer::SetSendBufferSize(int connection_id, int32 size) {
+  HttpConnection* connection = FindConnection(connection_id);
   if (connection == NULL)
     return;
+  connection->write_buf()->set_total_size_limit(size);
+}
 
-  connection->recv_data_.append(data, len);
-  while (connection->recv_data_.length()) {
-    if (connection->web_socket_.get()) {
+void HttpServer::DoAcceptLoop() {
+  int rv;
+  do {
+    rv = server_socket_->Accept(&accepted_socket_,
+                                base::Bind(&HttpServer::OnAcceptCompleted,
+                                           base::Unretained(this)));
+    if (rv == ERR_IO_PENDING)
+      return;
+    rv = DidAccept(rv);
+  } while (rv == OK);
+}
+
+void HttpServer::OnAcceptCompleted(int rv) {
+  if (DidAccept(rv) == OK)
+    DoAcceptLoop();
+}
+
+int HttpServer::DidAccept(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;
+  DoReadLoop(connection);
+  return OK;
+}
+
+void HttpServer::DoReadLoop(HttpConnection* connection) {
+  int rv;
+  do {
+    HttpConnection::ReadIOBuffer* read_buf = connection->read_buf();
+    // Increases read buffer size if necessary.
+    if (read_buf->GetUnusedCapacity() == 0 && !read_buf->IncreaseCapacity()) {
+      Close(connection->id());
+      return;
+    }
+
+    rv = connection->socket()->Read(
+        read_buf->GetUnusedIOBuffer(),
+        read_buf->GetUnusedCapacity(),
+        base::Bind(&HttpServer::OnReadCompleted,
+                   base::Unretained(this), connection->id()));
+    if (rv == ERR_IO_PENDING)
+      return;
+    rv = DidRead(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;
+
+  if (DidRead(connection, rv) == OK)
+    DoReadLoop(connection);
+}
+
+int HttpServer::DidRead(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->GetUnconsumedSize() > 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;
 
       if (result == WebSocket::FRAME_CLOSE ||
           result == WebSocket::FRAME_ERROR) {
         Close(connection->id());
         break;
       }
       delegate_->OnWebSocketMessage(connection->id(), message);
       continue;
     }
 
     HttpServerRequestInfo request;
     size_t pos = 0;
-    if (!ParseHeaders(connection, &request, &pos))
+    if (!ParseHeaders(read_buf->data(), read_buf->GetUnconsumedSize(),
+                      &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->GetUnconsumedSize() - pos < content_length)
         break;  // Not enough data was received yet.
-      request.data = connection->recv_data_.substr(pos, content_length);
+      request.data.assign(read_buf->data() + pos, content_length);
       pos += content_length;
     }
 
+    read_buf->DidConsume(pos);
     delegate_->OnHttpRequest(connection->id(), request);
-    connection->Shift(pos);
+  }
+
+  return OK;
+}
+
+void HttpServer::DoWriteLoop(HttpConnection* connection) {
+  int rv = OK;
+  HttpConnection::PendingWriteIOBuffer* 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,
+                   base::Unretained(this), connection->id()));
+    if (rv == ERR_IO_PENDING || rv == OK)
+      return;
+    rv = DidWrite(connection, rv);
   }
 }
 
-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::OnWriteCompleted(int connection_id, int rv) {
+  HttpConnection* connection = FindConnection(connection_id);
+  if (!connection)  // It might be closed right before by read error.
+    return;
+
+  if (DidWrite(connection, rv) == OK)
+    DoWriteLoop(connection);
 }
 
-HttpServer::~HttpServer() {
-  STLDeleteContainerPairSecondPointers(
-      id_to_connection_.begin(), id_to_connection_.end());
+int HttpServer::DidWrite(HttpConnection* connection, int rv) {
+  if (rv < 0) {
+    Close(connection->id());
+    return rv;
+  }
+
+  connection->write_buf()->DidConsume(rv);
+  return OK;
 }
 
+namespace {
+
 //
 // HTTP Request Parser
 // This HTTP request parser uses a simple state machine to quickly parse
 // through the headers.  The parser is not 100% complete, as it is designed
 // for use in this simple test driver.
 //
 // Known issues:
 //   - does not handle whitespace on first HTTP line correctly.  Expects
 //     a single space between the method/url and url/protocol.
 
@@ skipping 43 matching lines @@
     case '\r':
       return INPUT_CR;
     case '\n':
       return INPUT_LF;
     case ':':
       return INPUT_COLON;
   }
   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];
 
     bool transition = (next_state != state);
     HttpServerRequestInfo::HeadersMap::iterator it;
     if (transition) {
       // Do any actions based on state transitions.
       switch (state) {
         case ST_METHOD:
           info->method = buffer;
@@ skipping 51 matching lines @@
   return false;
 }
 
 HttpConnection* HttpServer::FindConnection(int connection_id) {
   IdToConnectionMap::iterator it = id_to_connection_.find(connection_id);
   if (it == id_to_connection_.end())
     return NULL;
   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