Merge 113138

Source/WebCore/Modules/websockets/WorkerThreadableWebSocketChannel.cpp

 /*
  * Copyright (C) 2011, 2012 Google Inc.  All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *     * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *     * Redistributions in binary form must reproduce the above
@@ skipping 39 matching lines @@
 #include <wtf/MainThread.h>
 #include <wtf/PassRefPtr.h>
 
 namespace WebCore {
 
 WorkerThreadableWebSocketChannel::WorkerThreadableWebSocketChannel(WorkerContext* context, WebSocketChannelClient* client, const String& taskMode)
     : m_workerContext(context)
     , m_workerClientWrapper(ThreadableWebSocketChannelClientWrapper::create(context, client))
     , m_bridge(Bridge::create(m_workerClientWrapper, m_workerContext, taskMode))
 {
+    m_bridge->initialize();
 }
 
 WorkerThreadableWebSocketChannel::~WorkerThreadableWebSocketChannel()
 {
     if (m_bridge)
         m_bridge->disconnect();
 }
 
 bool WorkerThreadableWebSocketChannel::useHixie76Protocol()
 {
@@ skipping 267 matching lines @@
     workerClientWrapper->didClose(unhandledBufferedAmount, closingHandshakeCompletion, code, reason);
 }
 
 void WorkerThreadableWebSocketChannel::Peer::didClose(unsigned long unhandledBufferedAmount, ClosingHandshakeCompletionStatus closingHandshakeCompletion, unsigned short code, const String& reason)
 {
     ASSERT(isMainThread());
     m_mainWebSocketChannel = 0;
     m_loaderProxy.postTaskForModeToWorkerContext(createCallbackTask(&workerContextDidClose, m_workerClientWrapper, unhandledBufferedAmount, closingHandshakeCompletion, code, reason), m_taskMode);
 }
 
-void WorkerThreadableWebSocketChannel::Bridge::setWebSocketChannel(ScriptExecutionContext* context, Bridge* thisPtr, Peer* peer, PassRefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, bool useHixie76Protocol)
-{
-    ASSERT_UNUSED(context, context->isWorkerContext());
-    thisPtr->m_peer = peer;
-    workerClientWrapper->setUseHixie76Protocol(useHixie76Protocol);
-    workerClientWrapper->setSyncMethodDone();
-}
-
-void WorkerThreadableWebSocketChannel::Bridge::mainThreadCreateWebSocketChannel(ScriptExecutionContext* context, Bridge* thisPtr, PassRefPtr<ThreadableWebSocketChannelClientWrapper> prpClientWrapper, const String& taskMode)
-{
-    ASSERT(isMainThread());
-    ASSERT_UNUSED(context, context->isDocument());
-
-    RefPtr<ThreadableWebSocketChannelClientWrapper> clientWrapper = prpClientWrapper;
-
-    Peer* peer = Peer::create(clientWrapper, thisPtr->m_loaderProxy, context, taskMode);
-    thisPtr->m_loaderProxy.postTaskForModeToWorkerContext(
-        createCallbackTask(&Bridge::setWebSocketChannel,
-                           AllowCrossThreadAccess(thisPtr),
-                           AllowCrossThreadAccess(peer), clientWrapper, peer->useHixie76Protocol()), taskMode);
-}
-
 WorkerThreadableWebSocketChannel::Bridge::Bridge(PassRefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper, PassRefPtr<WorkerContext> workerContext, const String& taskMode)
     : m_workerClientWrapper(workerClientWrapper)
     , m_workerContext(workerContext)
     , m_loaderProxy(m_workerContext->thread()->workerLoaderProxy())
     , m_taskMode(taskMode)
     , m_peer(0)
 {
     ASSERT(m_workerClientWrapper.get());
-    setMethodNotCompleted();
-    m_loaderProxy.postTaskToLoader(
-        createCallbackTask(&Bridge::mainThreadCreateWebSocketChannel,
-                           AllowCrossThreadAccess(this), m_workerClientWrapper, m_taskMode));
-    waitForMethodCompletion();
-    ASSERT(m_peer);
 }
 
 WorkerThreadableWebSocketChannel::Bridge::~Bridge()
 {
     disconnect();
 }
 
+class WorkerContextDidInitializeTask : public ScriptExecutionContext::Task {
+public:
+    static PassOwnPtr<ScriptExecutionContext::Task> create(WorkerThreadableWebSocketChannel::Peer* peer,
+                                                           PassRefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper,
+                                                           bool useHixie76Protocol)
+    {
+        return adoptPtr(new WorkerContextDidInitializeTask(peer, workerClientWrapper, useHixie76Protocol));
+    }
+
+    virtual ~WorkerContextDidInitializeTask() { }
+    virtual void performTask(ScriptExecutionContext* context) OVERRIDE
+    {
+        ASSERT_UNUSED(context, context->isWorkerContext());
+        m_workerClientWrapper->didCreateWebSocketChannel(m_peer, m_useHixie76Protocol);
+    }
+    virtual bool isCleanupTask() const OVERRIDE { return true; }
+
+private:
+    WorkerContextDidInitializeTask(WorkerThreadableWebSocketChannel::Peer* peer,
+                                   PassRefPtr<ThreadableWebSocketChannelClientWrapper> workerClientWrapper,
+                                   bool useHixie76Protocol)
+        : m_peer(peer)
+        , m_workerClientWrapper(workerClientWrapper)
+        , m_useHixie76Protocol(useHixie76Protocol)
+    {
+    }
+
+    WorkerThreadableWebSocketChannel::Peer* m_peer;
+    RefPtr<ThreadableWebSocketChannelClientWrapper> m_workerClientWrapper;
+    bool m_useHixie76Protocol;
+};
+
+void WorkerThreadableWebSocketChannel::Bridge::mainThreadInitialize(ScriptExecutionContext* context, WorkerLoaderProxy* loaderProxy, PassRefPtr<ThreadableWebSocketChannelClientWrapper> prpClientWrapper, const String& taskMode)
+{
+    ASSERT(isMainThread());
+    ASSERT_UNUSED(context, context->isDocument());
+
+    RefPtr<ThreadableWebSocketChannelClientWrapper> clientWrapper = prpClientWrapper;
+
+    Peer* peer = Peer::create(clientWrapper, *loaderProxy, context, taskMode);
+    bool sent = loaderProxy->postTaskForModeToWorkerContext(
+        WorkerContextDidInitializeTask::create(peer, clientWrapper, peer->useHixie76Protocol()), taskMode);
+    if (!sent) {
+        clientWrapper->clearPeer();
+        delete peer;
+    }
+}
+
+void WorkerThreadableWebSocketChannel::Bridge::initialize()
+{
+    ASSERT(!m_peer);
+    setMethodNotCompleted();
+    RefPtr<Bridge> protect(this);
+    m_loaderProxy.postTaskToLoader(
+        createCallbackTask(&Bridge::mainThreadInitialize,
+                           AllowCrossThreadAccess(&m_loaderProxy), m_workerClientWrapper, m_taskMode));
+    waitForMethodCompletion();
+    // m_peer may be null when the nested runloop exited before a peer has created.
+    m_peer = m_workerClientWrapper->peer();
+}
+
 void WorkerThreadableWebSocketChannel::mainThreadConnect(ScriptExecutionContext* context, Peer* peer, const KURL& url, const String& protocol)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->connect(url, protocol);
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::connect(const KURL& url, const String& protocol)
 {
     ASSERT(m_workerClientWrapper);
-    ASSERT(m_peer);
+    if (!m_peer)
+        return;
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadConnect, AllowCrossThreadAccess(m_peer), url, protocol));
 }
 
 void WorkerThreadableWebSocketChannel::mainThreadSend(ScriptExecutionContext* context, Peer* peer, const String& message)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->send(message);
@@ skipping 14 matching lines @@
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     RefPtr<Blob> blob = Blob::create(url, type, size);
     peer->send(*blob);
 }
 
 ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(const String& message)
 {
-    if (!m_workerClientWrapper)
+    if (!m_workerClientWrapper || !m_peer)
         return ThreadableWebSocketChannel::SendFail;
-    ASSERT(m_peer);
     setMethodNotCompleted();
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSend, AllowCrossThreadAccess(m_peer), message));
     RefPtr<Bridge> protect(this);
     waitForMethodCompletion();
     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();
     if (!clientWrapper)
         return ThreadableWebSocketChannel::SendFail;
     return clientWrapper->sendRequestResult();
 }
 
 ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(const ArrayBuffer& binaryData)
 {
-    if (!m_workerClientWrapper)
+    if (!m_workerClientWrapper || !m_peer)
         return ThreadableWebSocketChannel::SendFail;
-    ASSERT(m_peer);
     // ArrayBuffer isn't thread-safe, hence the content of ArrayBuffer is copied into Vector<char>.
     OwnPtr<Vector<char> > data = adoptPtr(new Vector<char>(binaryData.byteLength()));
     if (binaryData.byteLength())
         memcpy(data->data(), binaryData.data(), binaryData.byteLength());
     setMethodNotCompleted();
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSendArrayBuffer, AllowCrossThreadAccess(m_peer), data.release()));
     RefPtr<Bridge> protect(this);
     waitForMethodCompletion();
     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();
     if (!clientWrapper)
         return ThreadableWebSocketChannel::SendFail;
     return clientWrapper->sendRequestResult();
 }
 
 ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(const Blob& binaryData)
 {
-    if (!m_workerClientWrapper)
+    if (!m_workerClientWrapper || !m_peer)
         return ThreadableWebSocketChannel::SendFail;
-    ASSERT(m_peer);
     setMethodNotCompleted();
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSendBlob, AllowCrossThreadAccess(m_peer), binaryData.url(), binaryData.type(), binaryData.size()));
     RefPtr<Bridge> protect(this);
     waitForMethodCompletion();
     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();
     if (!clientWrapper)
         return ThreadableWebSocketChannel::SendFail;
     return clientWrapper->sendRequestResult();
 }
 
 void WorkerThreadableWebSocketChannel::mainThreadBufferedAmount(ScriptExecutionContext* context, Peer* peer)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->bufferedAmount();
 }
 
 unsigned long WorkerThreadableWebSocketChannel::Bridge::bufferedAmount()
 {
-    if (!m_workerClientWrapper)
+    if (!m_workerClientWrapper || !m_peer)
         return 0;
-    ASSERT(m_peer);
     setMethodNotCompleted();
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadBufferedAmount, AllowCrossThreadAccess(m_peer)));
     RefPtr<Bridge> protect(this);
     waitForMethodCompletion();
     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();
     if (clientWrapper)
         return clientWrapper->bufferedAmount();
     return 0;
 }
 
-void WorkerThreadableWebSocketChannel::mainThreadClose(ScriptExecutionContext* context, Peer* peer, int code, const String&reason)
+void WorkerThreadableWebSocketChannel::mainThreadClose(ScriptExecutionContext* context, Peer* peer, int code, const String& reason)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->close(code, reason);
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::close(int code, const String& reason)
 {
-    ASSERT(m_peer);
+    if (!m_peer)
+        return;
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadClose, AllowCrossThreadAccess(m_peer), code, reason));
 }
 
 void WorkerThreadableWebSocketChannel::mainThreadFail(ScriptExecutionContext* context, Peer* peer, const String& reason)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->fail(reason);
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::fail(const String& reason)
 {
-    ASSERT(m_peer);
+    if (!m_peer)
+        return;
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadFail, AllowCrossThreadAccess(m_peer), reason));
 }
 
 void WorkerThreadableWebSocketChannel::mainThreadDestroy(ScriptExecutionContext* context, Peer* peer)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     delete peer;
@@ skipping 14 matching lines @@
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->suspend();
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::suspend()
 {
-    ASSERT(m_peer);
+    if (!m_peer)
+        return;
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadSuspend, AllowCrossThreadAccess(m_peer)));
 }
 
 void WorkerThreadableWebSocketChannel::mainThreadResume(ScriptExecutionContext* context, Peer* peer)
 {
     ASSERT(isMainThread());
     ASSERT_UNUSED(context, context->isDocument());
     ASSERT(peer);
 
     peer->resume();
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::resume()
 {
-    ASSERT(m_peer);
+    if (!m_peer)
+        return;
     m_loaderProxy.postTaskToLoader(createCallbackTask(&WorkerThreadableWebSocketChannel::mainThreadResume, AllowCrossThreadAccess(m_peer)));
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::clearClientWrapper()
 {
     m_workerClientWrapper->clearClient();
 }
 
 void WorkerThreadableWebSocketChannel::Bridge::setMethodNotCompleted()
 {
@@ skipping 12 matching lines @@
     ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.get();
     while (m_workerContext && clientWrapper && !clientWrapper->syncMethodDone() && result != MessageQueueTerminated) {
         result = runLoop.runInMode(m_workerContext.get(), m_taskMode); // May cause this bridge to get disconnected, which makes m_workerContext become null.
         clientWrapper = m_workerClientWrapper.get();
     }
 }
 
 } // namespace WebCore
 
 #endif // ENABLE(WEB_SOCKETS)