WebSocket implementation in Chromium.

chrome/renderer/socket_stream_dispatcher.cc

Index: chrome/renderer/socket_stream_dispatcher.cc
diff --git a/chrome/renderer/socket_stream_dispatcher.cc b/chrome/renderer/socket_stream_dispatcher.cc
new file mode 100644
index 0000000000000000000000000000000000000000..33e81de45a1417baf2037f9c9df517cccc09c15e
--- /dev/null
+++ b/chrome/renderer/socket_stream_dispatcher.cc
@@ -0,0 +1,206 @@
+// Copyright 2009 Google Inc. All Rights Reserved.
+// Author: ukai@google.com (Fumitoshi Ukai)
+
+#include "chrome/renderer/socket_stream_dispatcher.h"
+
+#include <vector>
+
+#include "base/id_map.h"
+#include "base/ref_counted.h"
+#include "chrome/common/render_messages.h"
+#include "chrome/common/net/socket_stream.h"
+#include "chrome/renderer/render_thread.h"
+#include "googleurl/src/gurl.h"
+#include "webkit/glue/websocketstreamhandle_bridge.h"
+#include "webkit/glue/websocketstreamhandle_delegate.h"
+
+// IPCWebSocketStreamHandleBridge is owned by each SocketStreamHandle.
+// It communites with the main browser process via SocketStreamDispatcher.
+class IPCWebSocketStreamHandleBridge
+    : public base::RefCounted<IPCWebSocketStreamHandleBridge>,
+      public webkit_glue::WebSocketStreamHandleBridge {
+ public:
+  IPCWebSocketStreamHandleBridge(
+      IPC::Message::Sender* sender,
+      WebKit::WebSocketStreamHandle* handle,
+      webkit_glue::WebSocketStreamHandleDelegate* delegate)
+      : socket_id_(chrome_common_net::kNoSocketId),
+        sender_(sender),
+        handle_(handle),
+        delegate_(delegate) {}
+  virtual ~IPCWebSocketStreamHandleBridge();
+
+  // Returns the handle having given id or NULL if there is no such handle.
+  static IPCWebSocketStreamHandleBridge* FromSocketId(int id);
+
+  // webkit_glue::WebSocketStreamHandleBridge methods.
+  virtual void Connect(const GURL& url);
+  virtual bool Send(const std::vector<char>& data);
+  virtual void Close();
+
+  // Called by SocketStreamDispatcher.
+  void OnConnected(int max_amount_send_allowed);
+  void OnSentData(int amount_sent);
+  void OnReceivedData(const std::vector<char>& data);
+  void OnClosed();
+
+ private:
+  void DoConnect(const GURL& url);
+  int socket_id_;
+
+  IPC::Message::Sender* sender_;
+  WebKit::WebSocketStreamHandle* handle_;
+  webkit_glue::WebSocketStreamHandleDelegate* delegate_;
+
+  static IDMap<IPCWebSocketStreamHandleBridge> all_bridges;
+};
+
+IDMap<IPCWebSocketStreamHandleBridge>
+IPCWebSocketStreamHandleBridge::all_bridges;
+
+/* static */
+IPCWebSocketStreamHandleBridge* IPCWebSocketStreamHandleBridge::FromSocketId(
+    int id) {
+  return all_bridges.Lookup(id);
+}
+
+IPCWebSocketStreamHandleBridge::~IPCWebSocketStreamHandleBridge() {
+  LOG(INFO) << "IPCWebSocketStreamHandleBridge destructor socket_id="
+            << socket_id_;
+  if (socket_id_ != chrome_common_net::kNoSocketId) {
+    sender_->Send(new ViewHostMsg_Close(socket_id_));
+    socket_id_ = chrome_common_net::kNoSocketId;
+  }
+}
+
+void IPCWebSocketStreamHandleBridge::Connect(const GURL& url) {
+  DCHECK(sender_);
+  LOG(INFO) << "Connect url=" << url;
+  MessageLoop::current()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(this, &IPCWebSocketStreamHandleBridge::DoConnect,
+                        url));
+}
+
+bool IPCWebSocketStreamHandleBridge::Send(
+    const std::vector<char>& data) {
+  LOG(INFO) << "Send data.size=" << data.size();
+  if (sender_->Send(new ViewHostMsg_SocketStream_SendData(socket_id_, data))) {
+    if (delegate_)
+      delegate_->WillSendData(handle_, &data[0], data.size());
+    return true;
+  }
+  return false;
+}
+
+void IPCWebSocketStreamHandleBridge::Close() {
+  LOG(INFO) << "Close socket_id" << socket_id_;
+  sender_->Send(new ViewHostMsg_SocketStream_Close(socket_id_));
+}
+
+void IPCWebSocketStreamHandleBridge::OnConnected(int max_pending_send_allowed) {
+  LOG(INFO) << "IPCWebSocketStreamHandleBridge::OnConnected socket_id="
+            << socket_id_;
+  if (delegate_)
+    delegate_->DidOpenStream(handle_, max_pending_send_allowed);
+}
+
+void IPCWebSocketStreamHandleBridge::OnSentData(int amount_sent) {
+  if (delegate_)
+    delegate_->DidSendData(handle_, amount_sent);
+}
+
+void IPCWebSocketStreamHandleBridge::OnReceivedData(
+    const std::vector<char>& data) {
+  if (delegate_)
+    delegate_->DidReceiveData(handle_, &data[0], data.size());
+}
+
+void IPCWebSocketStreamHandleBridge::OnClosed() {
+  LOG(INFO) << "IPCWebSocketStreamHandleBridge::OnClosed";
+  if (socket_id_ != chrome_common_net::kNoSocketId) {
+    all_bridges.Remove(socket_id_);
+    socket_id_ = chrome_common_net::kNoSocketId;
+  }
+  if (delegate_) {
+    delegate_->DidClose(handle_);
+  }
+  delegate_ = NULL;
+  Release();
+}
+
+void IPCWebSocketStreamHandleBridge::DoConnect(const GURL& url) {
+  DCHECK(sender_);
+  DCHECK_EQ(socket_id_, chrome_common_net::kNoSocketId);
+  if (delegate_)
+    delegate_->WillOpenStream(handle_, url);
+
+  socket_id_ = all_bridges.Add(this);
+  DCHECK_NE(socket_id_, chrome_common_net::kNoSocketId);
+  if (sender_->Send(new ViewHostMsg_SocketStream_Connect(url, socket_id_))) {
+    LOG(INFO) << "Connect socket_id=" << socket_id_;
+    AddRef();  // Released in OnClosed().
+    // TODO(ukai): timeout to OnConnected.
+  } else {
+    LOG(INFO) << "Connect failed?";
+    DCHECK_EQ(socket_id_, chrome_common_net::kNoSocketId);
+    OnClosed();
+  }
+}
+
+SocketStreamDispatcher::SocketStreamDispatcher() {
+}
+
+/* static */
+webkit_glue::WebSocketStreamHandleBridge*
+SocketStreamDispatcher::CreateBridge(
+    WebKit::WebSocketStreamHandle* handle,
+    webkit_glue::WebSocketStreamHandleDelegate* delegate) {
+    return new IPCWebSocketStreamHandleBridge(
+        ChildThread::current(), handle, delegate);
+}
+
+bool SocketStreamDispatcher::OnMessageReceived(const IPC::Message& msg) {
+  LOG(INFO) << "SocketStreamDispatcher::OnMessageReceived";
+  bool handled = true;
+  IPC_BEGIN_MESSAGE_MAP(SocketStreamDispatcher, msg)
+    IPC_MESSAGE_HANDLER(ViewMsg_SocketStream_Connected, OnConnected)
+    IPC_MESSAGE_HANDLER(ViewMsg_SocketStream_SentData, OnSentData)
+    IPC_MESSAGE_HANDLER(ViewMsg_SocketStream_ReceivedData, OnReceivedData)
+    IPC_MESSAGE_HANDLER(ViewMsg_SocketStream_Closed, OnClosed)
+    IPC_MESSAGE_UNHANDLED(handled = false)
+  IPC_END_MESSAGE_MAP()
+  return handled;
+}
+
+void SocketStreamDispatcher::OnConnected(int socket_id,
+                                         int max_pending_send_allowed) {
+  LOG(INFO) << "SocketStreamDispatcher::OnConnected socket_id=" << socket_id
+            << " max_pending_send_allowed=" << max_pending_send_allowed;
+  IPCWebSocketStreamHandleBridge* bridge =
+      IPCWebSocketStreamHandleBridge::FromSocketId(socket_id);
+  if (bridge)
+    bridge->OnConnected(max_pending_send_allowed);
+}
+
+void SocketStreamDispatcher::OnSentData(int socket_id, int amount_sent) {
+  IPCWebSocketStreamHandleBridge* bridge =
+      IPCWebSocketStreamHandleBridge::FromSocketId(socket_id);
+  if (bridge)
+    bridge->OnSentData(amount_sent);
+}
+
+void SocketStreamDispatcher::OnReceivedData(
+    int socket_id, const std::vector<char>& data) {
+  IPCWebSocketStreamHandleBridge* bridge =
+      IPCWebSocketStreamHandleBridge::FromSocketId(socket_id);
+  if (bridge)
+    bridge->OnReceivedData(data);
+}
+
+void SocketStreamDispatcher::OnClosed(int socket_id) {
+  IPCWebSocketStreamHandleBridge* bridge =
+      IPCWebSocketStreamHandleBridge::FromSocketId(socket_id);
+  if (bridge)
+    bridge->OnClosed();
+}