1 | /* |
2 | * Copyright (C) 2011, 2012 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 | #include "WorkerThreadableWebSocketChannel.h" |
33 | |
34 | #include "Blob.h" |
35 | #include "Document.h" |
36 | #include "ScriptExecutionContext.h" |
37 | #include "SocketProvider.h" |
38 | #include "ThreadableWebSocketChannelClientWrapper.h" |
39 | #include "WebSocketChannel.h" |
40 | #include "WebSocketChannelClient.h" |
41 | #include "WorkerGlobalScope.h" |
42 | #include "WorkerLoaderProxy.h" |
43 | #include "WorkerRunLoop.h" |
44 | #include "WorkerThread.h" |
45 | #include <JavaScriptCore/ArrayBuffer.h> |
46 | #include <wtf/MainThread.h> |
47 | #include <wtf/text/WTFString.h> |
48 | |
49 | namespace WebCore { |
50 | |
51 | WorkerThreadableWebSocketChannel::WorkerThreadableWebSocketChannel(WorkerGlobalScope& context, WebSocketChannelClient& client, const String& taskMode, SocketProvider& provider) |
52 | : m_workerGlobalScope(context) |
53 | , m_workerClientWrapper(ThreadableWebSocketChannelClientWrapper::create(context, client)) |
54 | , m_bridge(Bridge::create(m_workerClientWrapper.copyRef(), m_workerGlobalScope.copyRef(), taskMode, provider)) |
55 | , m_socketProvider(provider) |
56 | { |
57 | m_bridge->initialize(); |
58 | } |
59 | |
60 | WorkerThreadableWebSocketChannel::~WorkerThreadableWebSocketChannel() |
61 | { |
62 | if (m_bridge) |
63 | m_bridge->disconnect(); |
64 | } |
65 | |
66 | WorkerThreadableWebSocketChannel::ConnectStatus WorkerThreadableWebSocketChannel::connect(const URL& url, const String& protocol) |
67 | { |
68 | if (m_bridge) |
69 | m_bridge->connect(url, protocol); |
70 | // connect is called asynchronously, so we do not have any possibility for synchronous errors. |
71 | return ConnectStatus::OK; |
72 | } |
73 | |
74 | String WorkerThreadableWebSocketChannel::subprotocol() |
75 | { |
76 | return m_workerClientWrapper->subprotocol(); |
77 | } |
78 | |
79 | String WorkerThreadableWebSocketChannel::extensions() |
80 | { |
81 | return m_workerClientWrapper->extensions(); |
82 | } |
83 | |
84 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::send(const String& message) |
85 | { |
86 | if (!m_bridge) |
87 | return ThreadableWebSocketChannel::SendFail; |
88 | return m_bridge->send(message); |
89 | } |
90 | |
91 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::send(const ArrayBuffer& binaryData, unsigned byteOffset, unsigned byteLength) |
92 | { |
93 | if (!m_bridge) |
94 | return ThreadableWebSocketChannel::SendFail; |
95 | return m_bridge->send(binaryData, byteOffset, byteLength); |
96 | } |
97 | |
98 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::send(Blob& binaryData) |
99 | { |
100 | if (!m_bridge) |
101 | return ThreadableWebSocketChannel::SendFail; |
102 | return m_bridge->send(binaryData); |
103 | } |
104 | |
105 | unsigned WorkerThreadableWebSocketChannel::bufferedAmount() const |
106 | { |
107 | if (!m_bridge) |
108 | return 0; |
109 | return m_bridge->bufferedAmount(); |
110 | } |
111 | |
112 | void WorkerThreadableWebSocketChannel::close(int code, const String& reason) |
113 | { |
114 | if (m_bridge) |
115 | m_bridge->close(code, reason); |
116 | } |
117 | |
118 | void WorkerThreadableWebSocketChannel::fail(const String& reason) |
119 | { |
120 | if (m_bridge) |
121 | m_bridge->fail(reason); |
122 | } |
123 | |
124 | void WorkerThreadableWebSocketChannel::disconnect() |
125 | { |
126 | m_bridge->disconnect(); |
127 | m_bridge = nullptr; |
128 | } |
129 | |
130 | void WorkerThreadableWebSocketChannel::suspend() |
131 | { |
132 | m_workerClientWrapper->suspend(); |
133 | if (m_bridge) |
134 | m_bridge->suspend(); |
135 | } |
136 | |
137 | void WorkerThreadableWebSocketChannel::resume() |
138 | { |
139 | m_workerClientWrapper->resume(); |
140 | if (m_bridge) |
141 | m_bridge->resume(); |
142 | } |
143 | |
144 | WorkerThreadableWebSocketChannel::Peer::Peer(Ref<ThreadableWebSocketChannelClientWrapper>&& clientWrapper, WorkerLoaderProxy& loaderProxy, ScriptExecutionContext& context, const String& taskMode, SocketProvider& provider) |
145 | : m_workerClientWrapper(WTFMove(clientWrapper)) |
146 | , m_loaderProxy(loaderProxy) |
147 | , m_mainWebSocketChannel(WebSocketChannel::create(downcast<Document>(context), *this, provider)) |
148 | , m_taskMode(taskMode) |
149 | { |
150 | ASSERT(isMainThread()); |
151 | } |
152 | |
153 | WorkerThreadableWebSocketChannel::Peer::~Peer() |
154 | { |
155 | ASSERT(isMainThread()); |
156 | if (m_mainWebSocketChannel) |
157 | m_mainWebSocketChannel->disconnect(); |
158 | } |
159 | |
160 | WorkerThreadableWebSocketChannel::ConnectStatus WorkerThreadableWebSocketChannel::Peer::connect(const URL& url, const String& protocol) |
161 | { |
162 | ASSERT(isMainThread()); |
163 | if (!m_mainWebSocketChannel) |
164 | return WorkerThreadableWebSocketChannel::ConnectStatus::KO; |
165 | return m_mainWebSocketChannel->connect(url, protocol); |
166 | } |
167 | |
168 | void WorkerThreadableWebSocketChannel::Peer::send(const String& message) |
169 | { |
170 | ASSERT(isMainThread()); |
171 | if (!m_mainWebSocketChannel) |
172 | return; |
173 | |
174 | ThreadableWebSocketChannel::SendResult sendRequestResult = m_mainWebSocketChannel->send(message); |
175 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), sendRequestResult](ScriptExecutionContext&) mutable { |
176 | workerClientWrapper->setSendRequestResult(sendRequestResult); |
177 | }, m_taskMode); |
178 | } |
179 | |
180 | void WorkerThreadableWebSocketChannel::Peer::send(const ArrayBuffer& binaryData) |
181 | { |
182 | ASSERT(isMainThread()); |
183 | if (!m_mainWebSocketChannel) |
184 | return; |
185 | |
186 | ThreadableWebSocketChannel::SendResult sendRequestResult = m_mainWebSocketChannel->send(binaryData, 0, binaryData.byteLength()); |
187 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), sendRequestResult](ScriptExecutionContext&) mutable { |
188 | workerClientWrapper->setSendRequestResult(sendRequestResult); |
189 | }, m_taskMode); |
190 | } |
191 | |
192 | void WorkerThreadableWebSocketChannel::Peer::send(Blob& binaryData) |
193 | { |
194 | ASSERT(isMainThread()); |
195 | if (!m_mainWebSocketChannel) |
196 | return; |
197 | |
198 | ThreadableWebSocketChannel::SendResult sendRequestResult = m_mainWebSocketChannel->send(binaryData); |
199 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), sendRequestResult](ScriptExecutionContext&) mutable { |
200 | workerClientWrapper->setSendRequestResult(sendRequestResult); |
201 | }, m_taskMode); |
202 | } |
203 | |
204 | void WorkerThreadableWebSocketChannel::Peer::bufferedAmount() |
205 | { |
206 | ASSERT(isMainThread()); |
207 | if (!m_mainWebSocketChannel) |
208 | return; |
209 | |
210 | unsigned bufferedAmount = m_mainWebSocketChannel->bufferedAmount(); |
211 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), bufferedAmount](ScriptExecutionContext& context) mutable { |
212 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
213 | workerClientWrapper->setBufferedAmount(bufferedAmount); |
214 | }, m_taskMode); |
215 | } |
216 | |
217 | void WorkerThreadableWebSocketChannel::Peer::close(int code, const String& reason) |
218 | { |
219 | ASSERT(isMainThread()); |
220 | if (!m_mainWebSocketChannel) |
221 | return; |
222 | m_mainWebSocketChannel->close(code, reason); |
223 | } |
224 | |
225 | void WorkerThreadableWebSocketChannel::Peer::fail(const String& reason) |
226 | { |
227 | ASSERT(isMainThread()); |
228 | if (!m_mainWebSocketChannel) |
229 | return; |
230 | m_mainWebSocketChannel->fail(reason); |
231 | } |
232 | |
233 | void WorkerThreadableWebSocketChannel::Peer::disconnect() |
234 | { |
235 | ASSERT(isMainThread()); |
236 | if (!m_mainWebSocketChannel) |
237 | return; |
238 | m_mainWebSocketChannel->disconnect(); |
239 | m_mainWebSocketChannel = nullptr; |
240 | } |
241 | |
242 | void WorkerThreadableWebSocketChannel::Peer::suspend() |
243 | { |
244 | ASSERT(isMainThread()); |
245 | if (!m_mainWebSocketChannel) |
246 | return; |
247 | m_mainWebSocketChannel->suspend(); |
248 | } |
249 | |
250 | void WorkerThreadableWebSocketChannel::Peer::resume() |
251 | { |
252 | ASSERT(isMainThread()); |
253 | if (!m_mainWebSocketChannel) |
254 | return; |
255 | m_mainWebSocketChannel->resume(); |
256 | } |
257 | |
258 | void WorkerThreadableWebSocketChannel::Peer::didConnect() |
259 | { |
260 | ASSERT(isMainThread()); |
261 | |
262 | String subprotocol = m_mainWebSocketChannel->subprotocol(); |
263 | String extensions = m_mainWebSocketChannel->extensions(); |
264 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), subprotocol = subprotocol.isolatedCopy(), extensions = extensions.isolatedCopy()](ScriptExecutionContext& context) mutable { |
265 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
266 | workerClientWrapper->setSubprotocol(subprotocol); |
267 | workerClientWrapper->setExtensions(extensions); |
268 | workerClientWrapper->didConnect(); |
269 | }, m_taskMode); |
270 | } |
271 | |
272 | void WorkerThreadableWebSocketChannel::Peer::didReceiveMessage(const String& message) |
273 | { |
274 | ASSERT(isMainThread()); |
275 | |
276 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), message = message.isolatedCopy()](ScriptExecutionContext& context) mutable { |
277 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
278 | workerClientWrapper->didReceiveMessage(message); |
279 | }, m_taskMode); |
280 | } |
281 | |
282 | void WorkerThreadableWebSocketChannel::Peer::didReceiveBinaryData(Vector<uint8_t>&& binaryData) |
283 | { |
284 | ASSERT(isMainThread()); |
285 | |
286 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), binaryData = WTFMove(binaryData)](ScriptExecutionContext& context) mutable { |
287 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
288 | workerClientWrapper->didReceiveBinaryData(WTFMove(binaryData)); |
289 | }, m_taskMode); |
290 | } |
291 | |
292 | void WorkerThreadableWebSocketChannel::Peer::didUpdateBufferedAmount(unsigned bufferedAmount) |
293 | { |
294 | ASSERT(isMainThread()); |
295 | |
296 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), bufferedAmount](ScriptExecutionContext& context) mutable { |
297 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
298 | workerClientWrapper->didUpdateBufferedAmount(bufferedAmount); |
299 | }, m_taskMode); |
300 | } |
301 | |
302 | void WorkerThreadableWebSocketChannel::Peer::didStartClosingHandshake() |
303 | { |
304 | ASSERT(isMainThread()); |
305 | |
306 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef()](ScriptExecutionContext& context) mutable { |
307 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
308 | workerClientWrapper->didStartClosingHandshake(); |
309 | }, m_taskMode); |
310 | } |
311 | |
312 | void WorkerThreadableWebSocketChannel::Peer::didClose(unsigned unhandledBufferedAmount, ClosingHandshakeCompletionStatus closingHandshakeCompletion, unsigned short code, const String& reason) |
313 | { |
314 | ASSERT(isMainThread()); |
315 | m_mainWebSocketChannel = nullptr; |
316 | |
317 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef(), unhandledBufferedAmount, closingHandshakeCompletion, code, reason = reason.isolatedCopy()](ScriptExecutionContext& context) mutable { |
318 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
319 | workerClientWrapper->didClose(unhandledBufferedAmount, closingHandshakeCompletion, code, reason); |
320 | }, m_taskMode); |
321 | } |
322 | |
323 | void WorkerThreadableWebSocketChannel::Peer::didReceiveMessageError() |
324 | { |
325 | ASSERT(isMainThread()); |
326 | |
327 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef()](ScriptExecutionContext& context) mutable { |
328 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
329 | workerClientWrapper->didReceiveMessageError(); |
330 | }, m_taskMode); |
331 | } |
332 | |
333 | void WorkerThreadableWebSocketChannel::Peer::didUpgradeURL() |
334 | { |
335 | ASSERT(isMainThread()); |
336 | |
337 | m_loaderProxy.postTaskForModeToWorkerGlobalScope([workerClientWrapper = m_workerClientWrapper.copyRef()](ScriptExecutionContext& context) mutable { |
338 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
339 | workerClientWrapper->didUpgradeURL(); |
340 | }, m_taskMode); |
341 | } |
342 | |
343 | WorkerThreadableWebSocketChannel::Bridge::Bridge(Ref<ThreadableWebSocketChannelClientWrapper>&& workerClientWrapper, Ref<WorkerGlobalScope>&& workerGlobalScope, const String& taskMode, Ref<SocketProvider>&& socketProvider) |
344 | : m_workerClientWrapper(WTFMove(workerClientWrapper)) |
345 | , m_workerGlobalScope(WTFMove(workerGlobalScope)) |
346 | , m_loaderProxy(m_workerGlobalScope->thread().workerLoaderProxy()) |
347 | , m_taskMode(taskMode) |
348 | , m_socketProvider(WTFMove(socketProvider)) |
349 | { |
350 | } |
351 | |
352 | WorkerThreadableWebSocketChannel::Bridge::~Bridge() |
353 | { |
354 | disconnect(); |
355 | } |
356 | |
357 | void WorkerThreadableWebSocketChannel::Bridge::mainThreadInitialize(ScriptExecutionContext& context, WorkerLoaderProxy& loaderProxy, Ref<ThreadableWebSocketChannelClientWrapper>&& clientWrapper, const String& taskMode, Ref<SocketProvider>&& provider) |
358 | { |
359 | ASSERT(isMainThread()); |
360 | ASSERT(context.isDocument()); |
361 | |
362 | bool sent = loaderProxy.postTaskForModeToWorkerGlobalScope({ |
363 | ScriptExecutionContext::Task::CleanupTask, |
364 | [clientWrapper = clientWrapper.copyRef(), &loaderProxy, peer = std::make_unique<Peer>(clientWrapper.copyRef(), loaderProxy, context, taskMode, WTFMove(provider))](ScriptExecutionContext& context) mutable { |
365 | ASSERT_UNUSED(context, context.isWorkerGlobalScope()); |
366 | if (clientWrapper->failedWebSocketChannelCreation()) { |
367 | // If Bridge::initialize() quitted earlier, we need to kick mainThreadDestroy() to delete the peer. |
368 | loaderProxy.postTaskToLoader([peer = WTFMove(peer)](ScriptExecutionContext& context) { |
369 | ASSERT(isMainThread()); |
370 | ASSERT_UNUSED(context, context.isDocument()); |
371 | }); |
372 | } else |
373 | clientWrapper->didCreateWebSocketChannel(peer.release()); |
374 | } |
375 | }, taskMode); |
376 | |
377 | if (!sent) |
378 | clientWrapper->clearPeer(); |
379 | } |
380 | |
381 | void WorkerThreadableWebSocketChannel::Bridge::initialize() |
382 | { |
383 | ASSERT(!m_peer); |
384 | setMethodNotCompleted(); |
385 | Ref<Bridge> protectedThis(*this); |
386 | |
387 | m_loaderProxy.postTaskToLoader([&loaderProxy = m_loaderProxy, workerClientWrapper = m_workerClientWrapper.copyRef(), taskMode = m_taskMode.isolatedCopy(), provider = m_socketProvider.copyRef()](ScriptExecutionContext& context) mutable { |
388 | mainThreadInitialize(context, loaderProxy, WTFMove(workerClientWrapper), taskMode, WTFMove(provider)); |
389 | }); |
390 | waitForMethodCompletion(); |
391 | |
392 | // m_peer may be null when the nested runloop exited before a peer has created. |
393 | m_peer = m_workerClientWrapper->peer(); |
394 | if (!m_peer) |
395 | m_workerClientWrapper->setFailedWebSocketChannelCreation(); |
396 | } |
397 | |
398 | void WorkerThreadableWebSocketChannel::Bridge::connect(const URL& url, const String& protocol) |
399 | { |
400 | if (!m_peer) |
401 | return; |
402 | |
403 | m_loaderProxy.postTaskToLoader([peer = m_peer, url = url.isolatedCopy(), protocol = protocol.isolatedCopy()](ScriptExecutionContext& context) { |
404 | ASSERT(isMainThread()); |
405 | ASSERT_UNUSED(context, context.isDocument()); |
406 | ASSERT(peer); |
407 | |
408 | if (peer->connect(url, protocol) == ThreadableWebSocketChannel::ConnectStatus::KO) |
409 | peer->didReceiveMessageError(); |
410 | }); |
411 | } |
412 | |
413 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(const String& message) |
414 | { |
415 | if (!m_peer) |
416 | return ThreadableWebSocketChannel::SendFail; |
417 | setMethodNotCompleted(); |
418 | |
419 | m_loaderProxy.postTaskToLoader([peer = m_peer, message = message.isolatedCopy()](ScriptExecutionContext& context) { |
420 | ASSERT(isMainThread()); |
421 | ASSERT_UNUSED(context, context.isDocument()); |
422 | ASSERT(peer); |
423 | |
424 | peer->send(message); |
425 | }); |
426 | |
427 | Ref<Bridge> protectedThis(*this); |
428 | waitForMethodCompletion(); |
429 | return m_workerClientWrapper->sendRequestResult(); |
430 | } |
431 | |
432 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(const ArrayBuffer& binaryData, unsigned byteOffset, unsigned byteLength) |
433 | { |
434 | if (!m_peer) |
435 | return ThreadableWebSocketChannel::SendFail; |
436 | |
437 | // ArrayBuffer isn't thread-safe, hence the content of ArrayBuffer is copied into Vector<char>. |
438 | Vector<char> data(byteLength); |
439 | if (binaryData.byteLength()) |
440 | memcpy(data.data(), static_cast<const char*>(binaryData.data()) + byteOffset, byteLength); |
441 | setMethodNotCompleted(); |
442 | |
443 | m_loaderProxy.postTaskToLoader([peer = m_peer, data = WTFMove(data)](ScriptExecutionContext& context) { |
444 | ASSERT(isMainThread()); |
445 | ASSERT_UNUSED(context, context.isDocument()); |
446 | ASSERT(peer); |
447 | |
448 | auto arrayBuffer = ArrayBuffer::create(data.data(), data.size()); |
449 | peer->send(arrayBuffer); |
450 | }); |
451 | |
452 | Ref<Bridge> protectedThis(*this); |
453 | waitForMethodCompletion(); |
454 | return m_workerClientWrapper->sendRequestResult(); |
455 | } |
456 | |
457 | ThreadableWebSocketChannel::SendResult WorkerThreadableWebSocketChannel::Bridge::send(Blob& binaryData) |
458 | { |
459 | if (!m_peer) |
460 | return ThreadableWebSocketChannel::SendFail; |
461 | setMethodNotCompleted(); |
462 | |
463 | m_loaderProxy.postTaskToLoader([peer = m_peer, url = binaryData.url().isolatedCopy(), type = binaryData.type().isolatedCopy(), size = binaryData.size()](ScriptExecutionContext& context) { |
464 | ASSERT(isMainThread()); |
465 | ASSERT_UNUSED(context, context.isDocument()); |
466 | ASSERT(peer); |
467 | |
468 | peer->send(Blob::deserialize(url, type, size, { })); |
469 | }); |
470 | |
471 | Ref<Bridge> protectedThis(*this); |
472 | waitForMethodCompletion(); |
473 | return m_workerClientWrapper->sendRequestResult(); |
474 | } |
475 | |
476 | unsigned WorkerThreadableWebSocketChannel::Bridge::bufferedAmount() |
477 | { |
478 | if (!m_peer) |
479 | return 0; |
480 | setMethodNotCompleted(); |
481 | |
482 | m_loaderProxy.postTaskToLoader([peer = m_peer](ScriptExecutionContext& context) { |
483 | ASSERT(isMainThread()); |
484 | ASSERT_UNUSED(context, context.isDocument()); |
485 | ASSERT(peer); |
486 | |
487 | peer->bufferedAmount(); |
488 | }); |
489 | |
490 | Ref<Bridge> protectedThis(*this); |
491 | waitForMethodCompletion(); |
492 | return m_workerClientWrapper->bufferedAmount(); |
493 | } |
494 | |
495 | void WorkerThreadableWebSocketChannel::Bridge::close(int code, const String& reason) |
496 | { |
497 | if (!m_peer) |
498 | return; |
499 | |
500 | m_loaderProxy.postTaskToLoader([peer = m_peer, code, reason = reason.isolatedCopy()](ScriptExecutionContext& context) { |
501 | ASSERT(isMainThread()); |
502 | ASSERT_UNUSED(context, context.isDocument()); |
503 | ASSERT(peer); |
504 | |
505 | peer->close(code, reason); |
506 | }); |
507 | } |
508 | |
509 | void WorkerThreadableWebSocketChannel::Bridge::fail(const String& reason) |
510 | { |
511 | if (!m_peer) |
512 | return; |
513 | |
514 | m_loaderProxy.postTaskToLoader([peer = m_peer, reason = reason.isolatedCopy()](ScriptExecutionContext& context) { |
515 | ASSERT(isMainThread()); |
516 | ASSERT_UNUSED(context, context.isDocument()); |
517 | ASSERT(peer); |
518 | |
519 | peer->fail(reason); |
520 | }); |
521 | } |
522 | |
523 | void WorkerThreadableWebSocketChannel::Bridge::disconnect() |
524 | { |
525 | clearClientWrapper(); |
526 | if (m_peer) { |
527 | m_loaderProxy.postTaskToLoader([peer = std::unique_ptr<Peer>(m_peer)](ScriptExecutionContext& context) { |
528 | ASSERT(isMainThread()); |
529 | ASSERT_UNUSED(context, context.isDocument()); |
530 | }); |
531 | m_peer = nullptr; |
532 | } |
533 | m_workerGlobalScope = nullptr; |
534 | } |
535 | |
536 | void WorkerThreadableWebSocketChannel::Bridge::suspend() |
537 | { |
538 | if (!m_peer) |
539 | return; |
540 | |
541 | m_loaderProxy.postTaskToLoader([peer = m_peer](ScriptExecutionContext& context) { |
542 | ASSERT(isMainThread()); |
543 | ASSERT_UNUSED(context, context.isDocument()); |
544 | ASSERT(peer); |
545 | |
546 | peer->suspend(); |
547 | }); |
548 | } |
549 | |
550 | void WorkerThreadableWebSocketChannel::Bridge::resume() |
551 | { |
552 | if (!m_peer) |
553 | return; |
554 | |
555 | m_loaderProxy.postTaskToLoader([peer = m_peer](ScriptExecutionContext& context) { |
556 | ASSERT(isMainThread()); |
557 | ASSERT_UNUSED(context, context.isDocument()); |
558 | ASSERT(peer); |
559 | |
560 | peer->resume(); |
561 | }); |
562 | } |
563 | |
564 | void WorkerThreadableWebSocketChannel::Bridge::clearClientWrapper() |
565 | { |
566 | m_workerClientWrapper->clearClient(); |
567 | } |
568 | |
569 | void WorkerThreadableWebSocketChannel::Bridge::setMethodNotCompleted() |
570 | { |
571 | m_workerClientWrapper->clearSyncMethodDone(); |
572 | } |
573 | |
574 | // Caller of this function should hold a reference to the bridge, because this function may call WebSocket::didClose() in the end, |
575 | // which causes the bridge to get disconnected from the WebSocket and deleted if there is no other reference. |
576 | void WorkerThreadableWebSocketChannel::Bridge::waitForMethodCompletion() |
577 | { |
578 | if (!m_workerGlobalScope) |
579 | return; |
580 | WorkerRunLoop& runLoop = m_workerGlobalScope->thread().runLoop(); |
581 | MessageQueueWaitResult result = MessageQueueMessageReceived; |
582 | ThreadableWebSocketChannelClientWrapper* clientWrapper = m_workerClientWrapper.ptr(); |
583 | while (m_workerGlobalScope && clientWrapper && !clientWrapper->syncMethodDone() && result != MessageQueueTerminated) { |
584 | result = runLoop.runInMode(m_workerGlobalScope.get(), m_taskMode); // May cause this bridge to get disconnected, which makes m_workerGlobalScope become null. |
585 | clientWrapper = m_workerClientWrapper.ptr(); |
586 | } |
587 | } |
588 | |
589 | } // namespace WebCore |
590 | |