1 /*

     2  * Copyright (C) 2009, 2010 Google Inc.  All rights reserved.

     3  *

     4  * Redistribution and use in source and binary forms, with or without

     5  * modification, are permitted provided that the following conditions are

     6  * met:

     7  *

     8  *     * Redistributions of source code must retain the above copyright

     9  * notice, this list of conditions and the following disclaimer.

    10  *     * Redistributions in binary form must reproduce the above

    11  * copyright notice, this list of conditions and the following disclaimer

    12  * in the documentation and/or other materials provided with the

    13  * distribution.

    14  *     * Neither the name of Google Inc. nor the names of its

    15  * contributors may be used to endorse or promote products derived from

    16  * this software without specific prior written permission.

    17  *

    18  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

    19  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

    20  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

    21  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT

    22  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

    23  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

    24  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

    25  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

    26  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

    27  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

    28  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

    29  */

    30 

    31 #include "config.h"

    32 

    33 #if ENABLE(WEB_SOCKETS) && ENABLE(WORKERS)

    34 

    35 #include "WorkerThreadableWebSocketChannel.h"

    36 

    37 #include "CrossThreadTask.h"

    38 #include "PlatformString.h"

    39 #include "ScriptExecutionContext.h"

    40 #include "ThreadableWebSocketChannelClientWrapper.h"

    41 #include "WebSocketChannel.h"

    42 #include "WebSocketChannelClient.h"

    43 #include "WorkerContext.h"

    44 #include "WorkerLoaderProxy.h"

    45 #include "WorkerRunLoop.h"

    46 #include "WorkerThread.h"

    47 

    48 #include <wtf/PassRefPtr.h>

    49 

    50 namespace WebCore {

    51 

    52 WorkerThreadableWebSocketChannel::WorkerThreadableWebSocketChannel(WorkerContext* context, WebSocketChannelClient* client, const String& taskMode, const KURL& url, const String& protocol)

    53     : m_workerContext(context)

    54     , m_workerClientWrapper(ThreadableWebSocketChannelClientWrapper::create(client))

    55     , m_bridge(Bridge::create(m_workerClientWrapper, m_workerContext, taskMode, url, protocol))

    56 {

    57 }

    58 

    59 WorkerThreadableWebSocketChannel::~WorkerThreadableWebSocketChannel()

    60 {

    61     if (m_bridge)

    62         m_bridge->disconnect();

    63 }

    64 

    65 void WorkerThreadableWebSocketChannel::connect()

    66 {

    67     if (m_bridge)

    68         m_bridge->connect();

    69 }

    70 

    71 bool WorkerThreadableWebSocketChannel::send(const String& message)

    72 {

    73     if (!m_bridge)

    74         return false;

    75     return m_bridge->send(message);

    76 }

    77 

    78 unsigned long WorkerThreadableWebSocketChannel::bufferedAmount() const

    79 {

    80     if (!m_bridge)

    81         return 0;

    82     return m_bridge->bufferedAmount();

    83 }

    84 

    85 void WorkerThreadableWebSocketChannel::close()

    86 {

    87     if (m_bridge)

    88         m_bridge->close();

    89 }

    90 

    91 void WorkerThreadableWebSocketChannel::disconnect()

    92 {

    93     m_bridge->disconnect();

    94     m_bridge.clear();

    95 }

    96 

    97 void WorkerThreadableWebSocketChannel::suspend()

    98 {

    99     m_workerClientWrapper->suspend();

   100     if (m_bridge)

   101         m_bridge->suspend();

   102 }

   103 

   104 void WorkerThreadableWebSocketChannel::resume()

   105 {

   106     m_workerClientWrapper->resume();

   107     if (m_bridge)

   108         m_bridge->resume();

   109 }

   110 

   111 WorkerThreadableWebSocketChannel::Peer::Peer(RefPtr<ThreadableWebSocketChannelClientWrapper> clientWrapper, WorkerLoaderProxy& loaderProxy, ScriptExecutionContext* context, const String& taskMode, const KURL& url, const String& protocol)

   112     : m_workerClientWrapper(clientWrapper)

   113     , m_loaderProxy(loaderProxy)

   114     , m_mainWebSocketChannel(WebSocketChannel::create(context, this, url, protocol))

   115     , m_taskMode(taskMode)

   116 {

   117     ASSERT(isMainThread());

   118 }

   119 

   120 WorkerThreadableWebSocketChannel::Peer::~Peer()

   121 {

   122     ASSERT(isMainThread());

   123     if (m_mainWebSocketChannel)

   124         m_mainWebSocketChannel->disconnect();

   125 }

   126 

   127 void WorkerThreadableWebSocketChannel::Peer::connect()

   128 {

   129     ASSERT(isMainThread());

   130     if (!m_mainWebSocketChannel)

   131         return;

   132     m_mainWebSocketChannel->connect();

   133 }

   134 

   135 static void workerContextDidSend(ScriptExecutionContext* context, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, bool sent)

   136 {

   137     ASSERT_UNUSED(context, context->isWorkerContext());

   138     workerClientWrapper->setSent(sent);

   139 }

   140 

   141 void WorkerThreadableWebSocketChannel::Peer::send(const String& message)

   142 {

   143     ASSERT(isMainThread());

   144     if (!m_mainWebSocketChannel || !m_workerClientWrapper)

   145         return;

   146     bool sent = m_mainWebSocketChannel->send(message);

   147     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidSend, m_workerClientWrapper, sent), m_taskMode);

   148 }

   149 

   150 static void workerContextDidGetBufferedAmount(ScriptExecutionContext* context, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, unsigned long bufferedAmount)

   151 {

   152     ASSERT_UNUSED(context, context->isWorkerContext());

   153     workerClientWrapper->setBufferedAmount(bufferedAmount);

   154 }

   155 

   156 void WorkerThreadableWebSocketChannel::Peer::bufferedAmount()

   157 {

   158     ASSERT(isMainThread());

   159     if (!m_mainWebSocketChannel || !m_workerClientWrapper)

   160         return;

   161     unsigned long bufferedAmount = m_mainWebSocketChannel->bufferedAmount();

   162     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidGetBufferedAmount, m_workerClientWrapper, bufferedAmount), m_taskMode);

   163 }

   164 

   165 void WorkerThreadableWebSocketChannel::Peer::close()

   166 {

   167     ASSERT(isMainThread());

   168     if (!m_mainWebSocketChannel)

   169         return;

   170     m_mainWebSocketChannel->close();

   171     m_mainWebSocketChannel = 0;

   172 }

   173 

   174 void WorkerThreadableWebSocketChannel::Peer::disconnect()

   175 {

   176     ASSERT(isMainThread());

   177     if (!m_mainWebSocketChannel)

   178         return;

   179     m_mainWebSocketChannel->disconnect();

   180     m_mainWebSocketChannel = 0;

   181 }

   182 

   183 void WorkerThreadableWebSocketChannel::Peer::suspend()

   184 {

   185     ASSERT(isMainThread());

   186     if (!m_mainWebSocketChannel)

   187         return;

   188     m_mainWebSocketChannel->suspend();

   189 }

   190 

   191 void WorkerThreadableWebSocketChannel::Peer::resume()

   192 {

   193     ASSERT(isMainThread());

   194     if (!m_mainWebSocketChannel)

   195         return;

   196     m_mainWebSocketChannel->resume();

   197 }

   198 

   199 static void workerContextDidConnect(ScriptExecutionContext* context, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper)

   200 {

   201     ASSERT_UNUSED(context, context->isWorkerContext());

   202     workerClientWrapper->didConnect();

   203 }

   204 

   205 void WorkerThreadableWebSocketChannel::Peer::didConnect()

   206 {

   207     ASSERT(isMainThread());

   208     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidConnect, m_workerClientWrapper), m_taskMode);

   209 }

   210 

   211 static void workerContextDidReceiveMessage(ScriptExecutionContext* context, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, const String& message)

   212 {

   213     ASSERT_UNUSED(context, context->isWorkerContext());

   214     workerClientWrapper->didReceiveMessage(message);

   215 }

   216 

   217 void WorkerThreadableWebSocketChannel::Peer::didReceiveMessage(const String& message)

   218 {

   219     ASSERT(isMainThread());

   220     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidReceiveMessage, m_workerClientWrapper, message), m_taskMode);

   221 }

   222 

   223 static void workerContextDidClose(ScriptExecutionContext* context, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, unsigned long unhandledBufferedAmount)

   224 {

   225     ASSERT_UNUSED(context, context->isWorkerContext());

   226     workerClientWrapper->didClose(unhandledBufferedAmount);

   227 }

   228 

   229 void WorkerThreadableWebSocketChannel::Peer::didClose(unsigned long unhandledBufferedAmount)

   230 {

   231     ASSERT(isMainThread());

   232     m_mainWebSocketChannel = 0;

   233     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidClose, m_workerClientWrapper, unhandledBufferedAmount), m_taskMode);

   234 }

   235 

   236 void WorkerThreadableWebSocketChannel::Bridge::setWebSocketChannel(ScriptExecutionContext* context, Bridge* thisPtr, Peer* peer, RefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper)

   237 {

   238     ASSERT_UNUSED(context, context->isWorkerContext());

   239     thisPtr->m_peer = peer;

   240     workerClientWrapper->setSyncMethodDone();

   241 }

   242 

   243 void WorkerThreadableWebSocketChannel::Bridge::mainThreadCreateWebSocketChannel(ScriptExecutionContext* context, Bridge* thisPtr, RefPtr<ThreadableWebSocketChannelClientWrapper> clientWrapper, const String& taskMode, const KURL& url, const String& protocol)

   244 {

   245     ASSERT(isMainThread());

   246     ASSERT_UNUSED(context, context->isDocument());

   247 

   248     Peer* peer = Peer::create(clientWrapper, thisPtr->m_loaderProxy, context, taskMode, url, protocol);

   249     thisPtr->m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&Bridge::setWebSocketChannel, thisPtr, peer, clientWrapper), taskMode);

   250 }

   251 

   252 WorkerThreadableWebSocketChannel::Bridge::Bridge(PassRefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, PassRefPtr<WorkerContext> workerContext, const String& taskMode, const KURL& url, const String& protocol)

   253     : m_workerClientWrapper(workerClientWrapper)

   254     , m_workerContext(workerContext)

   255     , m_loaderProxy(m_workerContext->thread()->workerLoaderProxy())

   256     , m_taskMode(taskMode)

   257     , m_peer(0)

   258 {

   259     ASSERT(m_workerClientWrapper.get());

   260     setMethodNotCompleted();

   261     m_loaderProxy.postTaskToLoader(createCallbackTask(&Bridge::mainThreadCreateWebSocketChannel, this, m_workerClientWrapper, m_taskMode, url, protocol));

   262     waitForMethodCompletion();

   263     ASSERT(m_peer);

   264 }

   265 

   266 WorkerThreadableWebSocketChannel::Bridge::~Bridge()

   267 {

   268     disconnect();

   269 }

   270 

   271 void WorkerThreadableWebSocketChannel::mainThreadConnect(ScriptExecutionContext* context, Peer* peer)

   272 {

   273     ASSERT(isMainThread());

   274     ASSERT_UNUSED(context, context->isDocument());

   275     ASSERT(peer);

   276 

   277     peer->connect();

   278 }

   279 

   280 void WorkerThreadableWebSocketChannel::Bridge::connect()

   281 {

   282     ASSERT(m_workerClientWrapper);

   283     ASSERT(m_peer);

   284     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadConnect, m_peer));

   285 }

   286 

   287 void WorkerThreadableWebSocketChannel::mainThreadSend(ScriptExecutionContext* context, Peer* peer, const String& message)

   288 {

   289     ASSERT(isMainThread());

   290     ASSERT_UNUSED(context, context->isDocument());

   291     ASSERT(peer);

   292 

   293     peer->send(message);

   294 }

   295 

   296 bool WorkerThreadableWebSocketChannel::Bridge::send(const String& message)

   297 {

   298     if (!m_workerClientWrapper)

   299         return false;

   300     ASSERT(m_peer);

   301     setMethodNotCompleted();

   302     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSend, m_peer, message));

   303     RefPtr<Bridge> protect(this);

   304     waitForMethodCompletion();

   305     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();

   306     return clientWrapper && clientWrapper->sent();

   307 }

   308 

   309 void WorkerThreadableWebSocketChannel::mainThreadBufferedAmount(ScriptExecutionContext* context, Peer* peer)

   310 {

   311     ASSERT(isMainThread());

   312     ASSERT_UNUSED(context, context->isDocument());

   313     ASSERT(peer);

   314 

   315     peer->bufferedAmount();

   316 }

   317 

   318 unsigned long WorkerThreadableWebSocketChannel::Bridge::bufferedAmount()

   319 {

   320     if (!m_workerClientWrapper)

   321         return 0;

   322     ASSERT(m_peer);

   323     setMethodNotCompleted();

   324     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadBufferedAmount, m_peer));

   325     RefPtr<Bridge> protect(this);

   326     waitForMethodCompletion();

   327     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();

   328     if (clientWrapper)

   329         return clientWrapper->bufferedAmount();

   330     return 0;

   331 }

   332 

   333 void WorkerThreadableWebSocketChannel::mainThreadClose(ScriptExecutionContext* context, Peer* peer)

   334 {

   335     ASSERT(isMainThread());

   336     ASSERT_UNUSED(context, context->isDocument());

   337     ASSERT(peer);

   338 

   339     peer->close();

   340 }

   341 

   342 void WorkerThreadableWebSocketChannel::Bridge::close()

   343 {

   344     ASSERT(m_peer);

   345     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadClose, m_peer));

   346 }

   347 

   348 void WorkerThreadableWebSocketChannel::mainThreadDestroy(ScriptExecutionContext* context, Peer* peer)

   349 {

   350     ASSERT(isMainThread());

   351     ASSERT_UNUSED(context, context->isDocument());

   352     ASSERT(peer);

   353 

   354     delete peer;

   355 }

   356 

   357 void WorkerThreadableWebSocketChannel::Bridge::disconnect()

   358 {

   359     clearClientWrapper();

   360     if (m_peer) {

   361         Peer* peer = m_peer;

   362         m_peer = 0;

   363         m_loaderProxy.postTaskToLoader(createCallbackTask(&mainThreadDestroy, peer));

   364     }

   365     m_workerContext = 0;

   366 }

   367 

   368 void WorkerThreadableWebSocketChannel::mainThreadSuspend(ScriptExecutionContext* context, Peer* peer)

   369 {

   370     ASSERT(isMainThread());

   371     ASSERT_UNUSED(context, context->isDocument());

   372     ASSERT(peer);

   373 

   374     peer->suspend();

   375 }

   376 

   377 void WorkerThreadableWebSocketChannel::Bridge::suspend()

   378 {

   379     ASSERT(m_peer);

   380     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSuspend, m_peer));

   381 }

   382 

   383 void WorkerThreadableWebSocketChannel::mainThreadResume(ScriptExecutionContext* context, Peer* peer)

   384 {

   385     ASSERT(isMainThread());

   386     ASSERT_UNUSED(context, context->isDocument());

   387     ASSERT(peer);

   388 

   389     peer->resume();

   390 }

   391 

   392 void WorkerThreadableWebSocketChannel::Bridge::resume()

   393 {

   394     ASSERT(m_peer);

   395     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadResume, m_peer));

   396 }

   397 

   398 void WorkerThreadableWebSocketChannel::Bridge::clearClientWrapper()

   399 {

   400     m_workerClientWrapper->clearClient();

   401 }

   402 

   403 void WorkerThreadableWebSocketChannel::Bridge::setMethodNotCompleted()

   404 {

   405     ASSERT(m_workerClientWrapper);

   406     m_workerClientWrapper->clearSyncMethodDone();

   407 }

   408 

   409 // Caller of this function should hold a reference to the bridge, because this function may call WebSocket::didClose() in the end,

   410 // which causes the bridge to get disconnected from the WebSocket and deleted if there is no other reference.

   411 void WorkerThreadableWebSocketChannel::Bridge::waitForMethodCompletion()

   412 {

   413     if (!m_workerContext)

   414         return;

   415     WorkerRunLoop& runLoop = m_workerContext->thread()->runLoop();

   416     MessageQueueWaitResult result = MessageQueueMessageReceived;

   417     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();

   418     while (m_workerContext && clientWrapper && !clientWrapper->syncMethodDone() && result != MessageQueueTerminated) {

   419         result = runLoop.runInMode(m_workerContext.get(), m_taskMode); // May cause this bridge to get disconnected, which makes m_workerContext become null.

   420         clientWrapper = m_workerClientWrapper.get();

   421     }

   422 }

   423 

   424 } // namespace WebCore

   425 

   426 #endif // ENABLE(WEB_SOCKETS)