mojo/services/network/http_connection_impl.cc

   1 // Copyright 2015 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "mojo/services/network/http_connection_impl.h"
   6
   7 #include <limits>
   8
   9 #include "base/bind.h"
  10 #include "base/bind_helpers.h"
  11 #include "base/callback.h"
  12 #include "base/logging.h"
  13 #include "base/stl_util.h"
  14 #include "base/strings/string_util.h"
  15 #include "mojo/common/handle_watcher.h"
  16 #include "mojo/services/network/http_server_impl.h"
  17 #include "mojo/services/network/net_adapters.h"
  18 #include "mojo/services/network/public/cpp/web_socket_read_queue.h"
  19 #include "mojo/services/network/public/cpp/web_socket_write_queue.h"
  20 #include "mojo/services/network/public/interfaces/web_socket.mojom.h"
  21 #include "net/base/net_errors.h"
  22 #include "net/http/http_request_headers.h"
  23 #include "net/http/http_status_code.h"
  24 #include "net/server/http_server.h"
  25 #include "net/server/http_server_request_info.h"
  26 #include "net/server/http_server_response_info.h"
  27 #include "third_party/mojo/src/mojo/public/cpp/bindings/type_converter.h"
  28 #include "third_party/mojo/src/mojo/public/cpp/system/data_pipe.h"
  29
  30 namespace mojo {
  31
  32 // SimpleDataPipeReader reads till end-of-file, stores the data in a string and
  33 // notifies completion.
  34 class HttpConnectionImpl::SimpleDataPipeReader {
  35  public:
  36   using CompletionCallback =
  37       base::Callback<void(SimpleDataPipeReader*, scoped_ptr<std::string>)>;
  38
  39   SimpleDataPipeReader() {}
  40   ~SimpleDataPipeReader() {}
  41
  42   void Start(ScopedDataPipeConsumerHandle consumer,
  43              const CompletionCallback& completion_callback) {
  44     DCHECK(consumer.is_valid() && !consumer_.is_valid());
  45     consumer_ = consumer.Pass();
  46     completion_callback_ = completion_callback;
  47     buffer_.reset(new std::string);
  48     ReadMore();
  49   }
  50
  51  private:
  52   void ReadMore() {
  53     const void* buf;
  54     uint32_t buf_size;
  55     MojoResult rv = BeginReadDataRaw(consumer_.get(), &buf, &buf_size,
  56                                      MOJO_READ_DATA_FLAG_NONE);
  57     if (rv == MOJO_RESULT_OK) {
  58       buffer_->append(static_cast<const char*>(buf), buf_size);
  59       EndReadDataRaw(consumer_.get(), buf_size);
  60       WaitToReadMore();
  61     } else if (rv == MOJO_RESULT_SHOULD_WAIT) {
  62       WaitToReadMore();
  63     } else if (rv == MOJO_RESULT_FAILED_PRECONDITION) {
  64       // We reached end-of-file.
  65       completion_callback_.Run(this, buffer_.Pass());
  66       // Note: This object may have been destroyed in the callback.
  67     } else {
  68       CHECK(false);
  69     }
  70   }
  71
  72   void WaitToReadMore() {
  73     watcher_.Start(consumer_.get(), MOJO_HANDLE_SIGNAL_READABLE,
  74                    MOJO_DEADLINE_INDEFINITE,
  75                    base::Bind(&SimpleDataPipeReader::OnHandleReady,
  76                               base::Unretained(this)));
  77   }
  78
  79   void OnHandleReady(MojoResult result) { ReadMore(); }
  80
  81   ScopedDataPipeConsumerHandle consumer_;
  82   common::HandleWatcher watcher_;
  83   CompletionCallback completion_callback_;
  84   scoped_ptr<std::string> buffer_;
  85
  86   DISALLOW_COPY_AND_ASSIGN(SimpleDataPipeReader);
  87 };
  88
  89 class HttpConnectionImpl::WebSocketImpl : public WebSocket,
  90                                           public ErrorHandler {
  91  public:
  92   // |connection| must outlive this object.
  93   WebSocketImpl(HttpConnectionImpl* connection,
  94                 InterfaceRequest<WebSocket> request,
  95                 ScopedDataPipeConsumerHandle send_stream,
  96                 WebSocketClientPtr client)
  97       : connection_(connection),
  98         binding_(this, request.Pass()),
  99         client_(client.Pass()),
 100         send_stream_(send_stream.Pass()),
 101         read_send_stream_(new WebSocketReadQueue(send_stream_.get())),
 102         pending_send_count_(0) {
 103     DCHECK(binding_.is_bound());
 104     DCHECK(client_);
 105     DCHECK(send_stream_.is_valid());
 106
 107     binding_.set_error_handler(this);
 108     client_.set_error_handler(this);
 109
 110     DataPipe data_pipe;
 111     receive_stream_ = data_pipe.producer_handle.Pass();
 112     write_receive_stream_.reset(new WebSocketWriteQueue(receive_stream_.get()));
 113
 114     client_->DidConnect("", "", data_pipe.consumer_handle.Pass());
 115   }
 116
 117   ~WebSocketImpl() override {}
 118
 119   void Close() {
 120     DCHECK(!IsClosing());
 121
 122     binding_.Close();
 123     client_.reset();
 124
 125     NotifyOwnerCloseIfAllDone();
 126   }
 127
 128   void OnReceivedWebSocketMessage(const std::string& data) {
 129     if (IsClosing())
 130       return;
 131
 132     // TODO(yzshen): It shouldn't be an issue to pass an empty message. However,
 133     // WebSocket{Read,Write}Queue doesn't handle that correctly.
 134     if (data.empty())
 135       return;
 136
 137     uint32_t size = static_cast<uint32_t>(data.size());
 138     write_receive_stream_->Write(
 139         &data[0], size,
 140         base::Bind(&WebSocketImpl::OnFinishedWritingReceiveStream,
 141                    base::Unretained(this), size));
 142   }
 143
 144  private:
 145   // WebSocket implementation.
 146   void Connect(const String& url,
 147                Array<String> protocols,
 148                const String& origin,
 149                ScopedDataPipeConsumerHandle send_stream,
 150                WebSocketClientPtr client) override {
 151     NOTREACHED();
 152   }
 153
 154   void Send(bool fin, MessageType type, uint32_t num_bytes) override {
 155     if (!fin || type != MESSAGE_TYPE_TEXT) {
 156       NOTIMPLEMENTED();
 157       Close();
 158     }
 159
 160     // TODO(yzshen): It shouldn't be an issue to pass an empty message. However,
 161     // WebSocket{Read,Write}Queue doesn't handle that correctly.
 162     if (num_bytes == 0)
 163       return;
 164
 165     pending_send_count_++;
 166     read_send_stream_->Read(
 167         num_bytes, base::Bind(&WebSocketImpl::OnFinishedReadingSendStream,
 168                               base::Unretained(this), num_bytes));
 169   }
 170
 171   void FlowControl(int64_t quota) override { NOTIMPLEMENTED(); }
 172
 173   void Close(uint16_t code, const String& reason) override {
 174     Close();
 175   }
 176
 177   // ErrorHandler implementation.
 178   void OnConnectionError() override { Close(); }
 179
 180   void OnFinishedReadingSendStream(uint32_t num_bytes, const char* data) {
 181     DCHECK_GT(pending_send_count_, 0u);
 182     pending_send_count_--;
 183
 184     if (data) {
 185       connection_->server_->server()->SendOverWebSocket(
 186           connection_->connection_id_, std::string(data, num_bytes));
 187     }
 188
 189     if (IsClosing())
 190       NotifyOwnerCloseIfAllDone();
 191   }
 192
 193   void OnFinishedWritingReceiveStream(uint32_t num_bytes, const char* buffer) {
 194     if (IsClosing())
 195       return;
 196
 197     if (buffer)
 198       client_->DidReceiveData(true, MESSAGE_TYPE_TEXT, num_bytes);
 199   }
 200
 201   // Checks whether Close() has been called.
 202   bool IsClosing() const { return !binding_.is_bound(); }
 203
 204   void NotifyOwnerCloseIfAllDone() {
 205     DCHECK(IsClosing());
 206
 207     if (pending_send_count_ == 0)
 208       connection_->OnWebSocketClosed();
 209   }
 210
 211   HttpConnectionImpl* const connection_;
 212
 213   Binding<WebSocket> binding_;
 214   WebSocketClientPtr client_;
 215
 216   ScopedDataPipeConsumerHandle send_stream_;
 217   scoped_ptr<WebSocketReadQueue> read_send_stream_;
 218   size_t pending_send_count_;
 219
 220   ScopedDataPipeProducerHandle receive_stream_;
 221   scoped_ptr<WebSocketWriteQueue> write_receive_stream_;
 222
 223   DISALLOW_COPY_AND_ASSIGN(WebSocketImpl);
 224 };
 225
 226 template <>
 227 struct TypeConverter<HttpRequestPtr, net::HttpServerRequestInfo> {
 228   static HttpRequestPtr Convert(const net::HttpServerRequestInfo& obj) {
 229     HttpRequestPtr request(HttpRequest::New());
 230     request->method = obj.method;
 231     request->url = obj.path;
 232     request->headers.resize(obj.headers.size());
 233     size_t index = 0;
 234     for (const auto& item : obj.headers) {
 235       HttpHeaderPtr header(HttpHeader::New());
 236       header->name = item.first;
 237       header->value = item.second;
 238       request->headers[index++] = header.Pass();
 239     }
 240     if (!obj.data.empty()) {
 241       uint32_t num_bytes = static_cast<uint32_t>(obj.data.size());
 242       MojoCreateDataPipeOptions options;
 243       options.struct_size = sizeof(MojoCreateDataPipeOptions);
 244       options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
 245       options.element_num_bytes = 1;
 246       options.capacity_num_bytes = num_bytes;
 247       DataPipe data_pipe(options);
 248       request->body = data_pipe.consumer_handle.Pass();
 249       MojoResult result =
 250           WriteDataRaw(data_pipe.producer_handle.get(), obj.data.data(),
 251                        &num_bytes, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE);
 252       CHECK_EQ(MOJO_RESULT_OK, result);
 253     }
 254     return request.Pass();
 255   }
 256 };
 257
 258 HttpConnectionImpl::HttpConnectionImpl(int connection_id,
 259                                        HttpServerImpl* server,
 260                                        HttpConnectionDelegatePtr delegate,
 261                                        HttpConnectionPtr* connection)
 262     : connection_id_(connection_id),
 263       server_(server),
 264       delegate_(delegate.Pass()),
 265       binding_(this, connection) {
 266   DCHECK(delegate_);
 267   binding_.set_error_handler(this);
 268   delegate_.set_error_handler(this);
 269 }
 270
 271 HttpConnectionImpl::~HttpConnectionImpl() {
 272   STLDeleteElements(&response_body_readers_);
 273 }
 274
 275 void HttpConnectionImpl::OnReceivedHttpRequest(
 276     const net::HttpServerRequestInfo& info) {
 277   if (IsClosing())
 278     return;
 279
 280   delegate_->OnReceivedRequest(
 281       HttpRequest::From(info), [this](HttpResponsePtr response) {
 282         if (response->body.is_valid()) {
 283           SimpleDataPipeReader* reader = new SimpleDataPipeReader;
 284           response_body_readers_.insert(reader);
 285           ScopedDataPipeConsumerHandle body = response->body.Pass();
 286           reader->Start(
 287               body.Pass(),
 288               base::Bind(&HttpConnectionImpl::OnFinishedReadingResponseBody,
 289                          base::Unretained(this), base::Passed(&response)));
 290         } else {
 291           OnFinishedReadingResponseBody(response.Pass(), nullptr, nullptr);
 292         }
 293       });
 294 }
 295
 296 void HttpConnectionImpl::OnReceivedWebSocketRequest(
 297     const net::HttpServerRequestInfo& info) {
 298   if (IsClosing())
 299     return;
 300
 301   delegate_->OnReceivedWebSocketRequest(
 302       HttpRequest::From(info),
 303       [this, info](InterfaceRequest<WebSocket> web_socket,
 304                    ScopedDataPipeConsumerHandle send_stream,
 305                    WebSocketClientPtr web_socket_client) {
 306         if (!web_socket.is_pending() || !send_stream.is_valid() ||
 307             !web_socket_client) {
 308           Close();
 309           return;
 310         }
 311
 312         web_socket_.reset(new WebSocketImpl(this, web_socket.Pass(),
 313                                             send_stream.Pass(),
 314                                             web_socket_client.Pass()));
 315         server_->server()->AcceptWebSocket(connection_id_, info);
 316       });
 317 }
 318
 319 void HttpConnectionImpl::OnReceivedWebSocketMessage(const std::string& data) {
 320   if (IsClosing())
 321     return;
 322
 323   web_socket_->OnReceivedWebSocketMessage(data);
 324 }
 325
 326 void HttpConnectionImpl::SetSendBufferSize(
 327     uint32_t size,
 328     const SetSendBufferSizeCallback& callback) {
 329   if (size > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()))
 330     size = std::numeric_limits<int32_t>::max();
 331
 332   server_->server()->SetSendBufferSize(connection_id_,
 333                                        static_cast<int32_t>(size));
 334   callback.Run(MakeNetworkError(net::OK));
 335 }
 336
 337 void HttpConnectionImpl::SetReceiveBufferSize(
 338     uint32_t size,
 339     const SetReceiveBufferSizeCallback& callback) {
 340   if (size > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()))
 341     size = std::numeric_limits<int32_t>::max();
 342
 343   server_->server()->SetReceiveBufferSize(connection_id_,
 344                                           static_cast<int32_t>(size));
 345   callback.Run(MakeNetworkError(net::OK));
 346 }
 347
 348 void HttpConnectionImpl::OnConnectionError() {
 349   // This method is called when the proxy side of |binding_| or the impl side of
 350   // |delegate_| has closed the pipe. Although it is set as error handler for
 351   // both |binding_| and |delegate_|, it will only be called at most once
 352   // because when called it closes/resets |binding_| and |delegate_|.
 353   Close();
 354 }
 355
 356 void HttpConnectionImpl::OnFinishedReadingResponseBody(
 357     HttpResponsePtr response,
 358     SimpleDataPipeReader* reader,
 359     scoped_ptr<std::string> body) {
 360   if (reader) {
 361     delete reader;
 362     response_body_readers_.erase(reader);
 363   }
 364
 365   net::HttpServerResponseInfo info(
 366       static_cast<net::HttpStatusCode>(response->status_code));
 367
 368   std::string content_type;
 369   for (size_t i = 0; i < response->headers.size(); ++i) {
 370     const HttpHeader& header = *(response->headers[i]);
 371
 372     if (body) {
 373       // net::HttpServerResponseInfo::SetBody() automatically sets
 374       // Content-Length and Content-Types, so skip the two here.
 375       //
 376       // TODO(yzshen): Consider adding to net::HttpServerResponseInfo a simple
 377       // setter for body which doesn't fiddle with headers.
 378       if (base::strcasecmp(header.name.data(),
 379                            net::HttpRequestHeaders::kContentLength) == 0) {
 380         continue;
 381       } else if (base::strcasecmp(header.name.data(),
 382                                   net::HttpRequestHeaders::kContentType) == 0) {
 383         content_type = header.value;
 384         continue;
 385       }
 386     }
 387     info.AddHeader(header.name, header.value);
 388   }
 389
 390   if (body)
 391     info.SetBody(*body, content_type);
 392
 393   server_->server()->SendResponse(connection_id_, info);
 394
 395   if (IsClosing())
 396     NotifyOwnerCloseIfAllDone();
 397 }
 398
 399 void HttpConnectionImpl::Close() {
 400   DCHECK(!IsClosing());
 401
 402   binding_.Close();
 403   delegate_.reset();
 404
 405   if (web_socket_)
 406     web_socket_->Close();
 407
 408   NotifyOwnerCloseIfAllDone();
 409 }
 410
 411 void HttpConnectionImpl::NotifyOwnerCloseIfAllDone() {
 412   DCHECK(IsClosing());
 413
 414   // Don't close the connection until all pending sends are done.
 415   bool should_wait = !response_body_readers_.empty() || web_socket_;
 416   if (!should_wait)
 417     server_->server()->Close(connection_id_);
 418 }
 419
 420 void HttpConnectionImpl::OnWebSocketClosed() {
 421   web_socket_.reset();
 422
 423   if (IsClosing()) {
 424     // The close operation is initiated by this object.
 425     NotifyOwnerCloseIfAllDone();
 426   } else {
 427     // The close operation is initiated by |web_socket_|; start closing this
 428     // object.
 429     Close();
 430   }
 431 }
 432
 433 }  // namespace mojo