src/chrome/browser/extensions/api/sockets_tcp_server/tcp_server_socket_event_dispatcher.cc

   1 // Copyright 2013 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 "chrome/browser/extensions/api/sockets_tcp_server/tcp_server_socket_event_dispatcher.h"
   6
   7 #include "chrome/browser/browser_process.h"
   8 #include "chrome/browser/extensions/api/socket/tcp_socket.h"
   9 #include "chrome/browser/profiles/profile.h"
  10 #include "chrome/browser/profiles/profile_manager.h"
  11 #include "extensions/browser/event_router.h"
  12 #include "extensions/browser/extension_system.h"
  13 #include "net/base/net_errors.h"
  14
  15 namespace extensions {
  16 namespace api {
  17
  18 using content::BrowserThread;
  19
  20 static
  21     base::LazyInstance<ProfileKeyedAPIFactory<TCPServerSocketEventDispatcher> >
  22     g_factory = LAZY_INSTANCE_INITIALIZER;
  23
  24 // static
  25 ProfileKeyedAPIFactory<TCPServerSocketEventDispatcher>*
  26     TCPServerSocketEventDispatcher::GetFactoryInstance() {
  27   return g_factory.Pointer();
  28 }
  29
  30 // static
  31 TCPServerSocketEventDispatcher* TCPServerSocketEventDispatcher::Get(
  32     Profile* profile) {
  33   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
  34
  35   return ProfileKeyedAPIFactory<TCPServerSocketEventDispatcher>::GetForProfile(
  36       profile);
  37 }
  38
  39 TCPServerSocketEventDispatcher::TCPServerSocketEventDispatcher(Profile* profile)
  40     : thread_id_(Socket::kThreadId),
  41       profile_(profile) {
  42   ApiResourceManager<ResumableTCPServerSocket>* server_manager =
  43       ApiResourceManager<ResumableTCPServerSocket>::Get(profile);
  44   DCHECK(server_manager) << "There is no server socket manager. "
  45     "If this assertion is failing during a test, then it is likely that "
  46     "TestExtensionSystem is failing to provide an instance of "
  47     "ApiResourceManager<ResumableTCPServerSocket>.";
  48   server_sockets_ = server_manager->data_;
  49
  50   ApiResourceManager<ResumableTCPSocket>* client_manager =
  51       ApiResourceManager<ResumableTCPSocket>::Get(profile);
  52   DCHECK(client_manager) << "There is no client socket manager. "
  53     "If this assertion is failing during a test, then it is likely that "
  54     "TestExtensionSystem is failing to provide an instance of "
  55     "ApiResourceManager<ResumableTCPSocket>.";
  56   client_sockets_ = client_manager->data_;
  57 }
  58
  59 TCPServerSocketEventDispatcher::~TCPServerSocketEventDispatcher() {}
  60
  61 TCPServerSocketEventDispatcher::AcceptParams::AcceptParams() {}
  62
  63 TCPServerSocketEventDispatcher::AcceptParams::~AcceptParams() {}
  64
  65 void TCPServerSocketEventDispatcher::OnServerSocketListen(
  66     const std::string& extension_id,
  67     int socket_id) {
  68   DCHECK(BrowserThread::CurrentlyOn(thread_id_));
  69
  70   StartSocketAccept(extension_id, socket_id);
  71 }
  72
  73 void TCPServerSocketEventDispatcher::OnServerSocketResume(
  74     const std::string& extension_id,
  75     int socket_id) {
  76   DCHECK(BrowserThread::CurrentlyOn(thread_id_));
  77
  78   StartSocketAccept(extension_id, socket_id);
  79 }
  80
  81 void TCPServerSocketEventDispatcher::StartSocketAccept(
  82     const std::string& extension_id,
  83     int socket_id) {
  84   DCHECK(BrowserThread::CurrentlyOn(thread_id_));
  85
  86   AcceptParams params;
  87   params.thread_id = thread_id_;
  88   params.profile_id = profile_;
  89   params.extension_id = extension_id;
  90   params.server_sockets = server_sockets_;
  91   params.client_sockets = client_sockets_;
  92   params.socket_id = socket_id;
  93
  94   StartAccept(params);
  95 }
  96
  97 // static
  98 void TCPServerSocketEventDispatcher::StartAccept(const AcceptParams& params) {
  99   DCHECK(BrowserThread::CurrentlyOn(params.thread_id));
 100
 101   ResumableTCPServerSocket* socket =
 102       params.server_sockets->Get(params.extension_id, params.socket_id);
 103   if (!socket) {
 104     // This can happen if the socket is closed while our callback is active.
 105     return;
 106   }
 107   DCHECK(params.extension_id == socket->owner_extension_id())
 108     << "Socket has wrong owner.";
 109
 110   // Don't start another accept if the socket has been paused.
 111   if (socket->paused())
 112     return;
 113
 114   socket->Accept(base::Bind(&TCPServerSocketEventDispatcher::AcceptCallback,
 115                  params));
 116 }
 117
 118 // static
 119 void TCPServerSocketEventDispatcher::AcceptCallback(
 120     const AcceptParams& params,
 121     int result_code,
 122     net::TCPClientSocket *socket) {
 123   DCHECK(BrowserThread::CurrentlyOn(params.thread_id));
 124
 125   if (result_code >= 0) {
 126     ResumableTCPSocket *client_socket =
 127         new ResumableTCPSocket(socket, params.extension_id, true);
 128     client_socket->set_paused(true);
 129     int client_socket_id = params.client_sockets->Add(client_socket);
 130
 131     // Dispatch "onAccept" event.
 132     sockets_tcp_server::AcceptInfo accept_info;
 133     accept_info.socket_id = params.socket_id;
 134     accept_info.client_socket_id = client_socket_id;
 135     scoped_ptr<base::ListValue> args =
 136         sockets_tcp_server::OnAccept::Create(accept_info);
 137     scoped_ptr<Event> event(
 138       new Event(sockets_tcp_server::OnAccept::kEventName, args.Pass()));
 139     PostEvent(params, event.Pass());
 140
 141     // Post a task to delay the "accept" until the socket is available, as
 142     // calling StartAccept at this point would error with ERR_IO_PENDING.
 143     BrowserThread::PostTask(
 144         params.thread_id, FROM_HERE,
 145         base::Bind(&TCPServerSocketEventDispatcher::StartAccept, params));
 146   } else {
 147     // Dispatch "onAcceptError" event but don't start another accept to avoid
 148     // potential infinite "accepts" if we have a persistent network error.
 149     sockets_tcp_server::AcceptErrorInfo accept_error_info;
 150     accept_error_info.socket_id = params.socket_id;
 151     accept_error_info.result_code = result_code;
 152     scoped_ptr<base::ListValue> args =
 153         sockets_tcp_server::OnAcceptError::Create(accept_error_info);
 154     scoped_ptr<Event> event(
 155       new Event(sockets_tcp_server::OnAcceptError::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     ResumableTCPServerSocket* socket =
 161         params.server_sockets->Get(params.extension_id, params.socket_id);
 162     if (socket) {
 163       socket->set_paused(true);
 164     }
 165   }
 166 }
 167
 168 // static
 169 void TCPServerSocketEventDispatcher::PostEvent(const AcceptParams& params,
 170                                                scoped_ptr<Event> event) {
 171   DCHECK(BrowserThread::CurrentlyOn(params.thread_id));
 172
 173   BrowserThread::PostTask(
 174       BrowserThread::UI, FROM_HERE,
 175       base::Bind(&DispatchEvent,
 176                  params.profile_id,
 177                  params.extension_id,
 178                  base::Passed(event.Pass())));
 179 }
 180
 181 // static
 182 void TCPServerSocketEventDispatcher::DispatchEvent(
 183     void* profile_id,
 184     const std::string& extension_id,
 185     scoped_ptr<Event> event) {
 186   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 187
 188   Profile* profile = reinterpret_cast<Profile*>(profile_id);
 189   if (!g_browser_process->profile_manager()->IsValidProfile(profile))
 190     return;
 191
 192   EventRouter* router = ExtensionSystem::Get(profile)->event_router();
 193   if (router)
 194     router->DispatchEventToExtension(extension_id, event.Pass());
 195 }
 196
 197 }  // namespace api
 198 }  // namespace extensions