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