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());
+  // 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) {
-  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::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);
   }
 }
 
-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;
 }
 
+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
 // 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