WorkerMessagingProxy.cpp [plain text]
#include "config.h"
#include "WorkerMessagingProxy.h"
#include "ContentSecurityPolicy.h"
#include "DOMWindow.h"
#include "DedicatedWorkerGlobalScope.h"
#include "DedicatedWorkerThread.h"
#include "Document.h"
#include "ErrorEvent.h"
#include "Event.h"
#include "EventNames.h"
#include "ExceptionCode.h"
#include "InspectorInstrumentation.h"
#include "MessageEvent.h"
#include "PageGroup.h"
#include "ScriptExecutionContext.h"
#include "Worker.h"
#include "WorkerDebuggerAgent.h"
#include "WorkerInspectorController.h"
#include <inspector/InspectorAgentBase.h>
#include <inspector/ScriptCallStack.h>
#include <runtime/ConsoleTypes.h>
#include <wtf/MainThread.h>
namespace WebCore {
WorkerGlobalScopeProxy* WorkerGlobalScopeProxy::create(Worker* worker)
{
return new WorkerMessagingProxy(worker);
}
WorkerMessagingProxy::WorkerMessagingProxy(Worker* workerObject)
: m_scriptExecutionContext(workerObject->scriptExecutionContext())
, m_workerObject(workerObject)
, m_mayBeDestroyed(false)
, m_unconfirmedMessageCount(0)
, m_workerThreadHadPendingActivity(false)
, m_askedToTerminate(false)
#if ENABLE(INSPECTOR)
, m_pageInspector(0)
#endif
{
ASSERT(m_workerObject);
ASSERT((m_scriptExecutionContext->isDocument() && isMainThread())
|| (m_scriptExecutionContext->isWorkerGlobalScope() && currentThread() == toWorkerGlobalScope(*m_scriptExecutionContext).thread().threadID()));
}
WorkerMessagingProxy::~WorkerMessagingProxy()
{
ASSERT(!m_workerObject);
ASSERT((m_scriptExecutionContext->isDocument() && isMainThread())
|| (m_scriptExecutionContext->isWorkerGlobalScope() && currentThread() == toWorkerGlobalScope(*m_scriptExecutionContext).thread().threadID()));
}
void WorkerMessagingProxy::startWorkerGlobalScope(const URL& scriptURL, const String& userAgent, const String& sourceCode, WorkerThreadStartMode startMode)
{
ASSERT_WITH_SECURITY_IMPLICATION(m_scriptExecutionContext->isDocument());
Document* document = static_cast<Document*>(m_scriptExecutionContext.get());
GroupSettings* settings = 0;
if (document->page())
settings = &document->page()->group().groupSettings();
RefPtr<DedicatedWorkerThread> thread = DedicatedWorkerThread::create(scriptURL, userAgent, settings, sourceCode, *this, *this, startMode, document->contentSecurityPolicy()->deprecatedHeader(), document->contentSecurityPolicy()->deprecatedHeaderType(), document->topOrigin());
workerThreadCreated(thread);
thread->start();
InspectorInstrumentation::didStartWorkerGlobalScope(m_scriptExecutionContext.get(), this, scriptURL);
}
void WorkerMessagingProxy::postMessageToWorkerObject(PassRefPtr<SerializedScriptValue> message, std::unique_ptr<MessagePortChannelArray> channels)
{
MessagePortChannelArray* channelsPtr = channels.release();
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext& context) {
Worker* workerObject = this->workerObject();
if (!workerObject || askedToTerminate())
return;
std::unique_ptr<MessagePortArray> ports = MessagePort::entanglePorts(context, std::unique_ptr<MessagePortChannelArray>(channelsPtr));
workerObject->dispatchEvent(MessageEvent::create(WTF::move(ports), message));
});
}
void WorkerMessagingProxy::postMessageToWorkerGlobalScope(PassRefPtr<SerializedScriptValue> message, std::unique_ptr<MessagePortChannelArray> channels)
{
if (m_askedToTerminate)
return;
MessagePortChannelArray* channelsPtr = channels.release();
ScriptExecutionContext::Task task([=] (ScriptExecutionContext& scriptContext) {
ASSERT_WITH_SECURITY_IMPLICATION(scriptContext.isWorkerGlobalScope());
DedicatedWorkerGlobalScope& context = static_cast<DedicatedWorkerGlobalScope&>(scriptContext);
std::unique_ptr<MessagePortArray> ports = MessagePort::entanglePorts(scriptContext, std::unique_ptr<MessagePortChannelArray>(channelsPtr));
context.dispatchEvent(MessageEvent::create(WTF::move(ports), message));
context.thread().workerObjectProxy().confirmMessageFromWorkerObject(context.hasPendingActivity());
});
if (m_workerThread) {
++m_unconfirmedMessageCount;
m_workerThread->runLoop().postTask(WTF::move(task));
} else
m_queuedEarlyTasks.append(std::make_unique<ScriptExecutionContext::Task>(WTF::move(task)));
}
void WorkerMessagingProxy::postTaskToLoader(ScriptExecutionContext::Task task)
{
ASSERT(m_scriptExecutionContext->isDocument());
m_scriptExecutionContext->postTask(WTF::move(task));
}
bool WorkerMessagingProxy::postTaskForModeToWorkerGlobalScope(ScriptExecutionContext::Task task, const String& mode)
{
if (m_askedToTerminate)
return false;
ASSERT(m_workerThread);
m_workerThread->runLoop().postTaskForMode(WTF::move(task), mode);
return true;
}
void WorkerMessagingProxy::postExceptionToWorkerObject(const String& errorMessage, int lineNumber, int columnNumber, const String& sourceURL)
{
String errorMessageCopy = errorMessage.isolatedCopy();
String sourceURLCopy = sourceURL.isolatedCopy();
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext& context) {
Worker* workerObject = this->workerObject();
if (!workerObject)
return;
bool errorHandled = !workerObject->dispatchEvent(ErrorEvent::create(errorMessageCopy, sourceURLCopy, lineNumber, columnNumber));
if (!errorHandled)
context.reportException(errorMessageCopy, lineNumber, columnNumber, sourceURLCopy, 0);
});
}
void WorkerMessagingProxy::postConsoleMessageToWorkerObject(MessageSource source, MessageLevel level, const String& message, int lineNumber, int columnNumber, const String& sourceURL)
{
String messageCopy = message.isolatedCopy();
String sourceURLCopy = sourceURL.isolatedCopy();
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext& context) {
if (askedToTerminate())
return;
context.addConsoleMessage(source, level, messageCopy, sourceURLCopy, lineNumber, columnNumber);
});
}
void WorkerMessagingProxy::workerThreadCreated(PassRefPtr<DedicatedWorkerThread> workerThread)
{
m_workerThread = workerThread;
if (m_askedToTerminate) {
m_workerThread->stop();
} else {
ASSERT(!m_unconfirmedMessageCount);
m_unconfirmedMessageCount = m_queuedEarlyTasks.size();
m_workerThreadHadPendingActivity = true;
auto queuedEarlyTasks = WTF::move(m_queuedEarlyTasks);
for (auto& task : queuedEarlyTasks)
m_workerThread->runLoop().postTask(WTF::move(*task));
}
}
void WorkerMessagingProxy::workerObjectDestroyed()
{
m_workerObject = 0;
m_scriptExecutionContext->postTask([this] (ScriptExecutionContext&) {
m_mayBeDestroyed = true;
if (m_workerThread)
terminateWorkerGlobalScope();
else
workerGlobalScopeDestroyedInternal();
});
}
void WorkerMessagingProxy::notifyNetworkStateChange(bool isOnline)
{
if (m_askedToTerminate)
return;
if (!m_workerThread)
return;
m_workerThread->runLoop().postTask([=] (ScriptExecutionContext& context) {
toWorkerGlobalScope(&context)->dispatchEvent(Event::create(isOnline ? eventNames().onlineEvent : eventNames().offlineEvent, false, false));
});
}
#if ENABLE(INSPECTOR)
void WorkerMessagingProxy::connectToInspector(WorkerGlobalScopeProxy::PageInspector* pageInspector)
{
if (m_askedToTerminate)
return;
ASSERT(!m_pageInspector);
m_pageInspector = pageInspector;
m_workerThread->runLoop().postTaskForMode([] (ScriptExecutionContext& context) {
toWorkerGlobalScope(&context)->workerInspectorController().connectFrontend();
}, WorkerDebuggerAgent::debuggerTaskMode);
}
void WorkerMessagingProxy::disconnectFromInspector()
{
m_pageInspector = 0;
if (m_askedToTerminate)
return;
m_workerThread->runLoop().postTaskForMode([] (ScriptExecutionContext& context) {
toWorkerGlobalScope(&context)->workerInspectorController().disconnectFrontend(Inspector::InspectorDisconnectReason::InspectorDestroyed);
}, WorkerDebuggerAgent::debuggerTaskMode);
}
void WorkerMessagingProxy::sendMessageToInspector(const String& message)
{
if (m_askedToTerminate)
return;
String messageCopy = message.isolatedCopy();
m_workerThread->runLoop().postTaskForMode([messageCopy] (ScriptExecutionContext& context) {
toWorkerGlobalScope(&context)->workerInspectorController().dispatchMessageFromFrontend(messageCopy);
}, WorkerDebuggerAgent::debuggerTaskMode);
WorkerDebuggerAgent::interruptAndDispatchInspectorCommands(m_workerThread.get());
}
#endif
void WorkerMessagingProxy::workerGlobalScopeDestroyed()
{
m_scriptExecutionContext->postTask([this] (ScriptExecutionContext&) {
workerGlobalScopeDestroyedInternal();
});
}
void WorkerMessagingProxy::workerGlobalScopeClosed()
{
m_scriptExecutionContext->postTask([this] (ScriptExecutionContext&) {
terminateWorkerGlobalScope();
});
}
void WorkerMessagingProxy::workerGlobalScopeDestroyedInternal()
{
m_askedToTerminate = true;
m_workerThread = 0;
InspectorInstrumentation::workerGlobalScopeTerminated(m_scriptExecutionContext.get(), this);
if (m_mayBeDestroyed)
delete this;
}
void WorkerMessagingProxy::terminateWorkerGlobalScope()
{
if (m_askedToTerminate)
return;
m_askedToTerminate = true;
if (m_workerThread)
m_workerThread->stop();
InspectorInstrumentation::workerGlobalScopeTerminated(m_scriptExecutionContext.get(), this);
}
#if ENABLE(INSPECTOR)
void WorkerMessagingProxy::postMessageToPageInspector(const String& message)
{
String messageCopy = message.isolatedCopy();
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext&) {
m_pageInspector->dispatchMessageFromWorker(messageCopy);
});
}
#endif
void WorkerMessagingProxy::confirmMessageFromWorkerObject(bool hasPendingActivity)
{
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext&) {
reportPendingActivityInternal(true, hasPendingActivity);
});
}
void WorkerMessagingProxy::reportPendingActivity(bool hasPendingActivity)
{
m_scriptExecutionContext->postTask([=] (ScriptExecutionContext&) {
reportPendingActivityInternal(false, hasPendingActivity);
});
}
void WorkerMessagingProxy::reportPendingActivityInternal(bool confirmingMessage, bool hasPendingActivity)
{
if (confirmingMessage && !m_askedToTerminate) {
ASSERT(m_unconfirmedMessageCount);
--m_unconfirmedMessageCount;
}
m_workerThreadHadPendingActivity = hasPendingActivity;
}
bool WorkerMessagingProxy::hasPendingActivity() const
{
return (m_unconfirmedMessageCount || m_workerThreadHadPendingActivity) && !m_askedToTerminate;
}
}