src/extensions/browser/api/sockets_tcp/tcp_socket_event_dispatcher.cc

   1 // Copyright 2014 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 "extensions/browser/api/sockets_tcp/tcp_socket_event_dispatcher.h"
   6
   7 #include "extensions/browser/api/socket/tcp_socket.h"
   8 #include "extensions/browser/event_router.h"
   9 #include "extensions/browser/extension_system.h"
  10 #include "extensions/browser/extensions_browser_client.h"
  11 #include "net/base/net_errors.h"
  12
  13 namespace {
  14 int kDefaultBufferSize = 4096;
  15 }
  16
  17 namespace extensions {
  18 namespace core_api {
  19
  20 using content::BrowserThread;
  21
  22 static base::LazyInstance<
  23     BrowserContextKeyedAPIFactory<TCPSocketEventDispatcher> > g_factory =
  24     LAZY_INSTANCE_INITIALIZER;
  25
  26 // static
  27 BrowserContextKeyedAPIFactory<TCPSocketEventDispatcher>*
  28 TCPSocketEventDispatcher::GetFactoryInstance() {
  29   return g_factory.Pointer();
  30 }
  31
  32 // static
  33 TCPSocketEventDispatcher* TCPSocketEventDispatcher::Get(
  34     content::BrowserContext* context) {
  35   DCHECK_CURRENTLY_ON(BrowserThread::UI);
  36
  37   return BrowserContextKeyedAPIFactory<TCPSocketEventDispatcher>::Get(context);
  38 }
  39
  40 TCPSocketEventDispatcher::TCPSocketEventDispatcher(
  41     content::BrowserContext* context)
  42     : thread_id_(Socket::kThreadId), browser_context_(context) {
  43   ApiResourceManager<ResumableTCPSocket>* manager =
  44       ApiResourceManager<ResumableTCPSocket>::Get(browser_context_);
  45   DCHECK(manager)
  46       << "There is no socket manager. "
  47          "If this assertion is failing during a test, then it is likely that "
  48          "TestExtensionSystem is failing to provide an instance of "
  49          "ApiResourceManager<ResumableTCPSocket>.";
  50   sockets_ = manager->data_;
  51 }
  52
  53 TCPSocketEventDispatcher::~TCPSocketEventDispatcher() {}
  54
  55 TCPSocketEventDispatcher::ReadParams::ReadParams() {}
  56
  57 TCPSocketEventDispatcher::ReadParams::~ReadParams() {}
  58
  59 void TCPSocketEventDispatcher::OnSocketConnect(const std::string& extension_id,
  60                                                int socket_id) {
  61   DCHECK_CURRENTLY_ON(thread_id_);
  62
  63   StartSocketRead(extension_id, socket_id);
  64 }
  65
  66 void TCPSocketEventDispatcher::OnSocketResume(const std::string& extension_id,
  67                                               int socket_id) {
  68   DCHECK_CURRENTLY_ON(thread_id_);
  69
  70   StartSocketRead(extension_id, socket_id);
  71 }
  72
  73 void TCPSocketEventDispatcher::StartSocketRead(const std::string& extension_id,
  74                                                int socket_id) {
  75   DCHECK_CURRENTLY_ON(thread_id_);
  76
  77   ReadParams params;
  78   params.thread_id = thread_id_;
  79   params.browser_context_id = browser_context_;
  80   params.extension_id = extension_id;
  81   params.sockets = sockets_;
  82   params.socket_id = socket_id;
  83
  84   StartRead(params);
  85 }
  86
  87 // static
  88 void TCPSocketEventDispatcher::StartRead(const ReadParams& params) {
  89   DCHECK_CURRENTLY_ON(params.thread_id);
  90
  91   ResumableTCPSocket* socket =
  92       params.sockets->Get(params.extension_id, params.socket_id);
  93   if (!socket) {
  94     // This can happen if the socket is closed while our callback is active.
  95     return;
  96   }
  97   DCHECK(params.extension_id == socket->owner_extension_id())
  98       << "Socket has wrong owner.";
  99
 100   // Don't start another read if the socket has been paused.
 101   if (socket->paused())
 102     return;
 103
 104   int buffer_size = socket->buffer_size();
 105   if (buffer_size <= 0)
 106     buffer_size = kDefaultBufferSize;
 107   socket->Read(buffer_size,
 108                base::Bind(&TCPSocketEventDispatcher::ReadCallback, params));
 109 }
 110
 111 // static
 112 void TCPSocketEventDispatcher::ReadCallback(
 113     const ReadParams& params,
 114     int bytes_read,
 115     scoped_refptr<net::IOBuffer> io_buffer) {
 116   DCHECK_CURRENTLY_ON(params.thread_id);
 117
 118   // If |bytes_read| == 0, the connection has been closed by the peer.
 119   // If |bytes_read| < 0, there was a network error, and |bytes_read| is a value
 120   // from "net::ERR_".
 121
 122   if (bytes_read == 0) {
 123     bytes_read = net::ERR_CONNECTION_CLOSED;
 124   }
 125
 126   if (bytes_read > 0) {
 127     // Dispatch "onReceive" event.
 128     sockets_tcp::ReceiveInfo receive_info;
 129     receive_info.socket_id = params.socket_id;
 130     receive_info.data = std::string(io_buffer->data(), bytes_read);
 131     scoped_ptr<base::ListValue> args =
 132         sockets_tcp::OnReceive::Create(receive_info);
 133     scoped_ptr<Event> event(
 134         new Event(sockets_tcp::OnReceive::kEventName, args.Pass()));
 135     PostEvent(params, event.Pass());
 136
 137     // Post a task to delay the read until the socket is available, as
 138     // calling StartReceive at this point would error with ERR_IO_PENDING.
 139     BrowserThread::PostTask(
 140         params.thread_id,
 141         FROM_HERE,
 142         base::Bind(&TCPSocketEventDispatcher::StartRead, params));
 143   } else if (bytes_read == net::ERR_IO_PENDING) {
 144     // This happens when resuming a socket which already had an
 145     // active "read" callback.
 146   } else {
 147     // Dispatch "onReceiveError" event but don't start another read to avoid
 148     // potential infinite reads if we have a persistent network error.
 149     sockets_tcp::ReceiveErrorInfo receive_error_info;
 150     receive_error_info.socket_id = params.socket_id;
 151     receive_error_info.result_code = bytes_read;
 152     scoped_ptr<base::ListValue> args =
 153         sockets_tcp::OnReceiveError::Create(receive_error_info);
 154     scoped_ptr<Event> event(
 155         new Event(sockets_tcp::OnReceiveError::kEventName, args.Pass()));
 156     PostEvent(params, event.Pass());
 157
 158     // Since we got an error, the socket is now "paused" until the application
 159     // "resumes" it.
 160     ResumableTCPSocket* socket =
 161         params.sockets->Get(params.extension_id, params.socket_id);
 162     if (socket) {
 163       socket->set_paused(true);
 164     }
 165   }
 166 }
 167
 168 // static
 169 void TCPSocketEventDispatcher::PostEvent(const ReadParams& params,
 170                                          scoped_ptr<Event> event) {
 171   DCHECK_CURRENTLY_ON(params.thread_id);
 172
 173   BrowserThread::PostTask(BrowserThread::UI,
 174                           FROM_HERE,
 175                           base::Bind(&DispatchEvent,
 176                                      params.browser_context_id,
 177                                      params.extension_id,
 178                                      base::Passed(event.Pass())));
 179 }
 180
 181 // static
 182 void TCPSocketEventDispatcher::DispatchEvent(void* browser_context_id,
 183                                              const std::string& extension_id,
 184                                              scoped_ptr<Event> event) {
 185   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 186
 187   content::BrowserContext* context =
 188       reinterpret_cast<content::BrowserContext*>(browser_context_id);
 189   if (!extensions::ExtensionsBrowserClient::Get()->IsValidContext(context))
 190     return;
 191
 192   EventRouter* event_router = EventRouter::Get(context);
 193   if (event_router)
 194     event_router->DispatchEventToExtension(extension_id, event.Pass());
 195 }
 196
 197 }  // namespace core_api
 198 }  // namespace extensions