Mojo: WebSocket interface now reuses the DataPipe for subsequent sends or

mojo/services/html_viewer/websockethandle_impl.cc

 // Copyright 2014 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 "mojo/services/html_viewer/websockethandle_impl.h"
 
 #include <vector>
 
+#include "base/bind.h"
+#include "base/memory/scoped_vector.h"
 #include "mojo/services/html_viewer/blink_basic_type_converters.h"
+#include "mojo/services/public/cpp/network/web_socket_read_queue.h"
+#include "mojo/services/public/cpp/network/web_socket_write_queue.h"
 #include "mojo/services/public/interfaces/network/network_service.mojom.h"
 #include "third_party/WebKit/public/platform/WebSerializedOrigin.h"
 #include "third_party/WebKit/public/platform/WebSocketHandleClient.h"
 #include "third_party/WebKit/public/platform/WebString.h"
 #include "third_party/WebKit/public/platform/WebURL.h"
 #include "third_party/WebKit/public/platform/WebVector.h"
 
 using blink::WebSerializedOrigin;
 using blink::WebSocketHandle;
 using blink::WebSocketHandleClient;
@@ skipping 40 matching lines @@
 // Blink WebSocketHandleClient interface.
 class WebSocketClientImpl : public InterfaceImpl<WebSocketClient> {
  public:
   explicit WebSocketClientImpl(WebSocketHandleImpl* handle,
                                blink::WebSocketHandleClient* client)
       : handle_(handle), client_(client) {}
   virtual ~WebSocketClientImpl() {}
 
  private:
   // WebSocketClient methods:
-  virtual void DidConnect(
-      bool fail,
-      const String& selected_subprotocol,
-      const String& extensions) OVERRIDE {
+  virtual void DidConnect(bool fail,
+                          const String& selected_subprotocol,
+                          const String& extensions,
+                          ScopedDataPipeConsumerHandle receive_stream)
+      OVERRIDE {
     blink::WebSocketHandleClient* client = client_;
     WebSocketHandleImpl* handle = handle_;
+    receive_stream_ = receive_stream.Pass();
+    read_queue_.reset(new WebSocketReadQueue(receive_stream_.get()));
     if (fail)
       handle->Disconnect();  // deletes |this|
     client->didConnect(handle,
                        fail,
                        selected_subprotocol.To<WebString>(),
                        extensions.To<WebString>());
     // |handle| can be deleted here.
   }
 
   virtual void DidReceiveData(bool fin,
                               WebSocket::MessageType type,
-                              ScopedDataPipeConsumerHandle data_pipe) OVERRIDE {
-    uint32_t num_bytes;
-    ReadDataRaw(data_pipe.get(), NULL, &num_bytes, MOJO_READ_DATA_FLAG_QUERY);
-    std::vector<char> data(num_bytes);
-    ReadDataRaw(
-        data_pipe.get(), &data[0], &num_bytes, MOJO_READ_DATA_FLAG_NONE);
-    const char* data_ptr = data.empty() ? NULL : &data[0];
-    client_->didReceiveData(handle_,
-                            fin,
-                            ConvertTo<WebSocketHandle::MessageType>(type),
-                            data_ptr,
-                            data.size());
-    // |handle| can be deleted here.
+                              uint32_t num_bytes) OVERRIDE {
+    read_queue_->Read(num_bytes,
+                      base::Bind(&WebSocketClientImpl::DidReadFromReceiveStream,
+                                 base::Unretained(this),
+                                 fin, type, num_bytes));
   }
 
   virtual void DidReceiveFlowControl(int64_t quota) OVERRIDE {
     client_->didReceiveFlowControl(handle_, quota);
     // |handle| can be deleted here.
   }
 
   virtual void DidFail(const String& message) OVERRIDE {
     blink::WebSocketHandleClient* client = client_;
     WebSocketHandleImpl* handle = handle_;
     handle->Disconnect();  // deletes |this|
     client->didFail(handle, message.To<WebString>());
     // |handle| can be deleted here.
   }
 
   virtual void DidClose(bool was_clean,
                         uint16_t code,
                         const String& reason) OVERRIDE {
     blink::WebSocketHandleClient* client = client_;
     WebSocketHandleImpl* handle = handle_;
     handle->Disconnect();  // deletes |this|
     client->didClose(handle, was_clean, code, reason.To<WebString>());
     // |handle| can be deleted here.
   }
 
+  void DidReadFromReceiveStream(bool fin,
+                                WebSocket::MessageType type,
+                                uint32_t num_bytes,
+                                const char* data) {
+    client_->didReceiveData(handle_,
+                            fin,
+                            ConvertTo<WebSocketHandle::MessageType>(type),
+                            data,
+                            num_bytes);
+    // |handle_| can be deleted here.
+  }
+
+  // |handle_| owns this object, so it is guaranteed to outlive us.
   WebSocketHandleImpl* handle_;
   blink::WebSocketHandleClient* client_;
+  ScopedDataPipeConsumerHandle receive_stream_;
+  scoped_ptr<WebSocketReadQueue> read_queue_;
 
   DISALLOW_COPY_AND_ASSIGN(WebSocketClientImpl);
 };
 
 WebSocketHandleImpl::WebSocketHandleImpl(NetworkService* network_service)
     : did_close_(false) {
   network_service->CreateWebSocket(Get(&web_socket_));
 }
 
 WebSocketHandleImpl::~WebSocketHandleImpl() {
   if (!did_close_) {
     // The connection is abruptly disconnected by the renderer without
     // closing handshake.
     web_socket_->Close(WebSocket::kAbnormalCloseCode, String());
   }
 }
 
 void WebSocketHandleImpl::connect(const WebURL& url,
                                   const WebVector<WebString>& protocols,
                                   const WebSerializedOrigin& origin,
                                   WebSocketHandleClient* client) {
   client_.reset(new WebSocketClientImpl(this, client));
   WebSocketClientPtr client_ptr;
   // TODO(mpcomplete): Is this the right ownership model? Or should mojo own
   // |client_|?
   WeakBindToProxy(client_.get(), &client_ptr);
+
+  DataPipe data_pipe;
+  send_stream_ = data_pipe.producer_handle.Pass();
+  write_queue_.reset(new WebSocketWriteQueue(send_stream_.get()));
   web_socket_->Connect(url.string().utf8(),
                        Array<String>::From(protocols),
                        origin.string().utf8(),
+                       data_pipe.consumer_handle.Pass(),
                        client_ptr.Pass());
 }
 
 void WebSocketHandleImpl::send(bool fin,
                                WebSocketHandle::MessageType type,
                                const char* data,
                                size_t size) {
   if (!client_)
     return;
 
-  // TODO(mpcomplete): reuse the data pipe for subsequent sends.
-  uint32_t num_bytes = static_cast<uint32_t>(size);
-  MojoCreateDataPipeOptions options;
-  options.struct_size = sizeof(MojoCreateDataPipeOptions);
-  options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
-  options.element_num_bytes = 1;
-  options.capacity_num_bytes = num_bytes;
-  DataPipe data_pipe(options);
-  WriteDataRaw(data_pipe.producer_handle.get(),
-               data,
-               &num_bytes,
-               MOJO_WRITE_DATA_FLAG_ALL_OR_NONE);
-  web_socket_->Send(
-      fin,
-      ConvertTo<WebSocket::MessageType>(type),
-      data_pipe.consumer_handle.Pass());
+  uint32_t size32 = static_cast<uint32_t>(size);
+  write_queue_->Write(
+      data, size32,
+      base::Bind(&WebSocketHandleImpl::DidWriteToSendStream,
+                 base::Unretained(this),
+                 fin, type, size32));
 }
 
 void WebSocketHandleImpl::flowControl(int64_t quota) {
   if (!client_)
     return;
 
   web_socket_->FlowControl(quota);
 }
 
 void WebSocketHandleImpl::close(unsigned short code, const WebString& reason) {
   web_socket_->Close(code, reason.utf8());
 }
 
+void WebSocketHandleImpl::DidWriteToSendStream(
+    bool fin,
+    WebSocketHandle::MessageType type,
+    uint32_t num_bytes,
+    const char* data) {
+  web_socket_->Send(fin, ConvertTo<WebSocket::MessageType>(type), num_bytes);
+}
+
 void WebSocketHandleImpl::Disconnect() {
   did_close_ = true;
   client_.reset();
 }
 
 }  // namespace mojo