WorkerMessagePortChannelProvider.cpp [plain text]
#include "config.h"
#include "WorkerMessagePortChannelProvider.h"
#include "MessagePort.h"
#include "WorkerGlobalScope.h"
#include "WorkerThread.h"
#include <wtf/MainThread.h>
#include <wtf/RunLoop.h>
namespace WebCore {
WorkerMessagePortChannelProvider::WorkerMessagePortChannelProvider(WorkerGlobalScope& scope)
: m_scope(scope)
{
}
WorkerMessagePortChannelProvider::~WorkerMessagePortChannelProvider()
{
while (!m_takeAllMessagesCallbacks.isEmpty()) {
auto first = m_takeAllMessagesCallbacks.begin();
first->value({ }, [] { });
m_takeAllMessagesCallbacks.remove(first);
}
while (!m_activityCallbacks.isEmpty()) {
auto first = m_activityCallbacks.begin();
first->value(HasActivity::No);
m_activityCallbacks.remove(first);
}
}
void WorkerMessagePortChannelProvider::createNewMessagePortChannel(const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
{
callOnMainThread([local, remote] {
MessagePortChannelProvider::singleton().createNewMessagePortChannel(local, remote);
});
}
void WorkerMessagePortChannelProvider::entangleLocalPortInThisProcessToRemote(const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
{
callOnMainThread([local, remote] {
MessagePortChannelProvider::singleton().entangleLocalPortInThisProcessToRemote(local, remote);
});
}
void WorkerMessagePortChannelProvider::messagePortDisentangled(const MessagePortIdentifier& local)
{
callOnMainThread([local] {
MessagePortChannelProvider::singleton().messagePortDisentangled(local);
});
}
void WorkerMessagePortChannelProvider::messagePortClosed(const MessagePortIdentifier&)
{
ASSERT_NOT_REACHED();
}
void WorkerMessagePortChannelProvider::postMessageToRemote(MessageWithMessagePorts&& message, const MessagePortIdentifier& remoteTarget)
{
callOnMainThread([message = WTFMove(message), remoteTarget]() mutable {
MessagePortChannelProvider::singleton().postMessageToRemote(WTFMove(message), remoteTarget);
});
}
void WorkerMessagePortChannelProvider::takeAllMessagesForPort(const MessagePortIdentifier& identifier, CompletionHandler<void(Vector<MessageWithMessagePorts>&&, Function<void()>&&)>&& callback)
{
uint64_t callbackIdentifier = ++m_lastCallbackIdentifier;
m_takeAllMessagesCallbacks.add(callbackIdentifier, WTFMove(callback));
callOnMainThread([this, workerThread = makeRef(m_scope.thread()), callbackIdentifier, identifier]() mutable {
MessagePortChannelProvider::singleton().takeAllMessagesForPort(identifier, [this, workerThread = WTFMove(workerThread), callbackIdentifier](Vector<MessageWithMessagePorts>&& messages, Function<void()>&& completionHandler) {
workerThread->runLoop().postTaskForMode([this, callbackIdentifier, messages = WTFMove(messages), completionHandler = WTFMove(completionHandler)](auto& scope) mutable {
ASSERT_UNUSED(scope, this == &downcast<WorkerGlobalScope>(scope).messagePortChannelProvider());
m_takeAllMessagesCallbacks.take(callbackIdentifier)(WTFMove(messages), [completionHandler = WTFMove(completionHandler)]() mutable {
callOnMainThread(WTFMove(completionHandler));
});
}, WorkerRunLoop::defaultMode());
});
});
}
void WorkerMessagePortChannelProvider::checkRemotePortForActivity(const MessagePortIdentifier& remoteTarget, CompletionHandler<void(HasActivity)>&& callback)
{
uint64_t callbackIdentifier = ++m_lastCallbackIdentifier;
m_activityCallbacks.add(callbackIdentifier, WTFMove(callback));
callOnMainThread([this, workerThread = makeRef(m_scope.thread()), callbackIdentifier, remoteTarget]() mutable {
MessagePortChannelProvider::singleton().checkRemotePortForActivity(remoteTarget, [this, workerThread = WTFMove(workerThread), callbackIdentifier](auto hasActivity) {
workerThread->runLoop().postTaskForMode([this, callbackIdentifier, hasActivity](auto& scope) mutable {
ASSERT_UNUSED(scope, this == &downcast<WorkerGlobalScope>(scope).messagePortChannelProvider());
m_activityCallbacks.take(callbackIdentifier)(hasActivity);
}, WorkerRunLoop::defaultMode());
});
});
}
}