[OBSOLETE][DO NOT SUBMIT][DO NOT COMMIT]] Browser-side implementation of WebSocket Blob receive.

content/browser/renderer_host/websocket_host.h

 // Copyright 2013 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.
 
 #ifndef CONTENT_BROWSER_RENDERER_HOST_WEBSOCKET_HOST_H_
 #define CONTENT_BROWSER_RENDERER_HOST_WEBSOCKET_HOST_H_
 
 #include <stdint.h>
-
+#include <queue>
 #include <string>
 #include <vector>
 
 #include "base/macros.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/time/time.h"
 #include "content/common/content_export.h"
 #include "content/common/websocket.h"
+#include "net/websockets/websocket_event_interface.h"
+#include "net/websockets/websocket_frame.h"
+#include "storage/browser/blob/blob_data_handle.h"
 
 class GURL;
 
 namespace url {
 class Origin;
 }  // namespace url
 
 namespace net {
 class WebSocketChannel;
 class URLRequestContext;
 }  // namespace net
 
 namespace IPC {
 class Message;
 }  // namespace IPC
 
 namespace content {
 
+class WebSocketBlobReceiver;
 class WebSocketBlobSender;
 class WebSocketDispatcherHost;
 
 // Host of net::WebSocketChannel. The lifetime of an instance of this class is
 // completely controlled by the WebSocketDispatcherHost object.
 class CONTENT_EXPORT WebSocketHost {
  public:
   WebSocketHost(int routing_id,
                 WebSocketDispatcherHost* dispatcher,
                 net::URLRequestContext* url_request_context,
                 base::TimeDelta delay);
   virtual ~WebSocketHost();
 
   // The renderer process is going away.
   // This function is virtual for testing.
   virtual void GoAway();
 
   // General message dispatch. WebSocketDispatcherHost::OnMessageReceived
   // delegates to this method after looking up the |routing_id|.
   virtual bool OnMessageReceived(const IPC::Message& message);
 
   int routing_id() const { return routing_id_; }
 
   bool handshake_succeeded() const { return handshake_succeeded_; }
   void OnHandshakeSucceeded() { handshake_succeeded_ = true; }
 
  private:
   class WebSocketEventHandler;
+  using ReceiveQuotaProvider = size_t;
+  using ReceiveQuotaConsumer = size_t;
+
+  // This class provides an abstraction for when there are multiple active
+  // providers of receive quota, but only one can actual consume receive quota
+  // (ie. receive messages) at any one time.
+  class ReceiveQuotaMultiplexer {
+   public:
+    ReceiveQuotaMultiplexer();
+
+    // Sets the channel It must be called before any of the other
+    // methods. SetChannel(nullptr) can be used to prevent any further quota
+    // from being supplied to the channel.
+    void SetChannel(net::WebSocketChannel* channel);
+
+    // Sets the current consumer. The pointer remains owned by the caller, and
+    // must remain valid until this method is called again, or the class is
+    // destroyed. The consumer must be initially set before any quota-related
+    // methods are called.
+    void SetConsumer(ReceiveQuotaConsumer* consumer);
+
+    // Adds quota for ReceiveQuotaProvider |provider|. The addition is performed
+    // by this method. If |provider| is also the current quota consumer, returns
+    // true.
+    bool AddQuota(ReceiveQuotaProvider* provider, size_t quota);
+
+    // Returns the quota available from the currently active quota consumer. In
+    // other words, how many bytes of data may be sent to the consumer right
+    // now.
+    size_t AvailableQuota() const;
+
+    // Marks some quota as consumed. Call immediately before sending some data
+    // to the current consumer.
+    void ConsumedQuota(size_t quota);
+
+    // Called when we receive a frame from the channel, so that we can track
+    // the channel's view of available quota correctly.
+    void ReceivedFrame(size_t size);
+
+    // Calls channel_->SendFlowControl() if appropriate.
+    void PublishMoreQuotaIfAvailable();
+
+   private:
+    net::WebSocketChannel* channel_ = nullptr;
+    size_t channel_quota_ = 0;
+    ReceiveQuotaConsumer* current_consumer_ = nullptr;
+
+    DISALLOW_COPY_AND_ASSIGN(ReceiveQuotaMultiplexer);
+  };
+
+  struct QueuedFrame {
+    QueuedFrame(bool fin,
+                net::WebSocketFrameHeader::OpCode type,
+                const std::vector<char>& data);
+    QueuedFrame(QueuedFrame&& rhs);
+    ~QueuedFrame();
+    QueuedFrame& operator=(QueuedFrame&& rhs);
+    bool fin = false;
+    net::WebSocketFrameHeader::OpCode type =
+        net::WebSocketFrameHeader::kOpCodeText;
+    std::vector<char> data;
+  };
+
+  struct DropChannelParameters;
+
+  class BlobReceiverClient;
 
   // Handlers for each message type, dispatched by OnMessageReceived(), as
   // defined in content/common/websocket_messages.h
 
   void OnAddChannelRequest(const GURL& socket_url,
                            const std::vector<std::string>& requested_protocols,
                            const url::Origin& origin,
                            int render_frame_id);
 
   void AddChannel(const GURL& socket_url,
                   const std::vector<std::string>& requested_protocols,
                   const url::Origin& origin,
                   int render_frame_id);
 
   void OnSendBlob(const std::string& uuid, uint64_t expected_size);
 
   void OnSendFrame(bool fin,
                    WebSocketMessageType type,
                    const std::vector<char>& data);
 
   void OnFlowControl(int64_t quota);
 
   void OnDropChannel(bool was_clean, uint16_t code, const std::string& reason);
 
+  void OnBinaryTypeChanged(WebSocketBinaryType new_type);
+
+  void OnBlobConfirmed();
+
   void BlobSendComplete(int result);
 
+  // Returns true if the message should be queued.
+  bool ShouldQueue(size_t data_size);
+
+  // Adds a message to the end of the queue.
+  void AppendToQueue(bool fin,
+                     net::WebSocketFrameHeader::OpCode type,
+                     const std::vector<char>& data);
+
+  // Attempts to send queued messages, and publish quota to WebSocketChannel if
+  // there is any left. May destroy |this| as a side-effect.
+  void FlushQueueAndPublishQuotaIfAvailable();
+
+  // Either passes the frame to the renderer or, if a Blob is being constructed,
+  // appends the data to the Blob. May start writing a new Blob, in which case
+  // |started_blob_receive| is set to true and the same frame will need to be
+  // processed again. In this case the caller has to ensure that the frame is
+  // queued, or, if it is already in the queue, not removed.
+  net::WebSocketEventInterface::ChannelState SendFrameInternal(
+      bool fin,
+      net::WebSocketFrameHeader::OpCode type,
+      const std::vector<char>& data,
+      bool* started_blob_receive);
+
+  // Starts receiving a blob.
+  void StartReceivingBlob(bool bin, const std::vector<char>& data);
+
+  // Finishes receiving a blob.
+  void FinishReceivingBlob(scoped_ptr<storage::BlobDataHandle> blob_data_handle,
+                           uint64_t size);
+
+  // Blob creation has failed. Kills the channel and destroys |this|.
+  void BlobReceiveFailed(int net_error_code);
+
+  // Returns true if a "DropChannel" message should be delayed.
+  bool ShouldDelayDropChannel() const;
+
+  void SetPendingDropChannel(bool was_clean,
+                             uint16_t code,
+                             const std::string& reason);
+
+  // Sends a DropChannel IPC, resulting in |this| being deleted.
+  void DoDelayedDropChannel();
+
   // non-NULL if and only if this object is currently in "blob sending mode".
   scoped_ptr<WebSocketBlobSender> blob_sender_;
 
   // The channel we use to send events to the network.
   scoped_ptr<net::WebSocketChannel> channel_;
 
   // The WebSocketHostDispatcher that created this object.
   WebSocketDispatcherHost* const dispatcher_;
 
   // The URL request context for the channel.
   net::URLRequestContext* const url_request_context_;
 
   // The ID used to route messages.
   const int routing_id_;
 
   // Delay used for per-renderer WebSocket throttling.
   base::TimeDelta delay_;
 
   // SendFlowControl() is delayed when OnFlowControl() is called before
   // AddChannel() is called.
   // Zero indicates there is no pending SendFlowControl().
   int64_t pending_flow_control_quota_;
 
   // handshake_succeeded_ is set and used by WebSocketDispatcherHost
   // to manage counters for per-renderer WebSocket throttling.
   bool handshake_succeeded_;
 
+  WebSocketBinaryType binary_type_;
+
+  std::queue<QueuedFrame> data_frame_queue_;
+  std::queue<scoped_ptr<storage::BlobDataHandle>> unconfirmed_blob_queue_;
+  ReceiveQuotaMultiplexer receive_quota_multiplexer_;
+  ReceiveQuotaProvider renderer_quota_;
+  ReceiveQuotaProvider blob_receiver_quota_;
+
+  scoped_ptr<WebSocketBlobReceiver> blob_receiver_;
+  scoped_ptr<DropChannelParameters> pending_drop_channel_;
+
   base::WeakPtrFactory<WebSocketHost> weak_ptr_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(WebSocketHost);
 };
 
 }  // namespace content
 
 #endif  // CONTENT_BROWSER_RENDERER_HOST_WEBSOCKET_HOST_H_