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,
+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(
     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)
+    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);
+  response.SetContentHeaders(data.size(), content_type);
   SendResponse(connection_id, response);
+  SendRaw(connection_id, data);
 }
 
 void HttpServer::Send200(int connection_id,
                          const std::string& data,
                          const std::string& content_type) {
   Send(connection_id, HTTP_OK, data, content_type);
 }
 
 void HttpServer::Send404(int connection_id) {
   SendResponse(connection_id, HttpServerResponseInfo::CreateFor404());
 }
 
 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.

[mmenke, 2014/08/14 16:36:02]

This seems really weird - this means we can still read from a "closed"
connection, and pass data along to the delegate after the consumer closed a
connection, and we tell the delegate the connection has been closed...  Or am I
missing something?

[byungchul, 2014/08/14 18:44:03]

Changed to do entire Close() in next run loop.

+  base::MessageLoopProxy::current()->DeleteSoon(FROM_HERE, 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);
+  DCHECK(connection);
+  connection->read_buf()->set_max_buffer_size(size);
 }
 
-void HttpServer::DidRead(StreamListenSocket* socket,
-                         const char* data,
-                         int len) {
-  HttpConnection* connection = FindConnection(socket);
-  DCHECK(connection != NULL);
-  if (connection == NULL)
+void HttpServer::SetSendBufferSize(int connection_id, int32 size) {
+  HttpConnection* connection = FindConnection(connection_id);
+  DCHECK(connection);
+  connection->write_buf()->set_max_buffer_size(size);
+}
+
+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::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;
+  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->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;
 
       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->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::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,
+                   weak_ptr_factory_.GetWeakPtr(), connection->id()));
+    if (rv == ERR_IO_PENDING || rv == OK)
+      return;
+    rv = HandleWriteResult(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 (HandleWriteResult(connection, rv) == OK)
+    DoWriteLoop(connection);
 }
 
-HttpServer::~HttpServer() {
-  STLDeleteContainerPairSecondPointers(
-      id_to_connection_.begin(), id_to_connection_.end());
+int HttpServer::HandleWriteResult(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