1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
/*
* Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines
import kotlinx.coroutines.internal.*
import org.w3c.dom.*
import kotlin.coroutines.*
import kotlin.js.Promise
private const val MAX_DELAY = Int.MAX_VALUE.toLong()
private fun delayToInt(timeMillis: Long): Int =
timeMillis.coerceIn(0, MAX_DELAY).toInt()
internal sealed class SetTimeoutBasedDispatcher: CoroutineDispatcher(), Delay {
inner class ScheduledMessageQueue : MessageQueue() {
internal val processQueue: dynamic = { process() }
override fun schedule() {
scheduleQueueProcessing()
}
override fun reschedule() {
setTimeout(processQueue, 0)
}
}
internal val messageQueue = ScheduledMessageQueue()
abstract fun scheduleQueueProcessing()
override fun dispatch(context: CoroutineContext, block: Runnable) {
messageQueue.enqueue(block)
}
override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
val handle = setTimeout({ block.run() }, delayToInt(timeMillis))
return ClearTimeout(handle)
}
override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
val handle = setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, delayToInt(timeMillis))
// Actually on cancellation, but clearTimeout is idempotent
continuation.invokeOnCancellation(handler = ClearTimeout(handle).asHandler)
}
}
internal object NodeDispatcher : SetTimeoutBasedDispatcher() {
override fun scheduleQueueProcessing() {
process.nextTick(messageQueue.processQueue)
}
}
internal object SetTimeoutDispatcher : SetTimeoutBasedDispatcher() {
override fun scheduleQueueProcessing() {
setTimeout(messageQueue.processQueue, 0)
}
}
private class ClearTimeout(private val handle: Int) : CancelHandler(), DisposableHandle {
override fun dispose() {
clearTimeout(handle)
}
override fun invoke(cause: Throwable?) {
dispose()
}
override fun toString(): String = "ClearTimeout[$handle]"
}
internal class WindowDispatcher(private val window: Window) : CoroutineDispatcher(), Delay {
private val queue = WindowMessageQueue(window)
override fun dispatch(context: CoroutineContext, block: Runnable) = queue.enqueue(block)
override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, delayToInt(timeMillis))
}
override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
val handle = window.setTimeout({ block.run() }, delayToInt(timeMillis))
return object : DisposableHandle {
override fun dispose() {
window.clearTimeout(handle)
}
}
}
}
private class WindowMessageQueue(private val window: Window) : MessageQueue() {
private val messageName = "dispatchCoroutine"
init {
window.addEventListener("message", { event: dynamic ->
if (event.source == window && event.data == messageName) {
event.stopPropagation()
process()
}
}, true)
}
override fun schedule() {
Promise.resolve(Unit).then({ process() })
}
override fun reschedule() {
window.postMessage(messageName, "*")
}
}
/**
* An abstraction over JS scheduling mechanism that leverages micro-batching of dispatched blocks without
* paying the cost of JS callbacks scheduling on every dispatch.
*
* Queue uses two scheduling mechanisms:
* 1) [schedule] is used to schedule the initial processing of the message queue.
* JS engine-specific microtask mechanism is used in order to boost performance on short runs and a dispatch batch
* 2) [reschedule] is used to schedule processing of the queue after yield to the JS event loop.
* JS engine-specific macrotask mechanism is used not to starve animations and non-coroutines macrotasks.
*
* Yet there could be a long tail of "slow" reschedules, but it should be amortized by the queue size.
*/
internal abstract class MessageQueue : ArrayQueue<Runnable>() {
val yieldEvery = 16 // yield to JS macrotask event loop after this many processed messages
private var scheduled = false
abstract fun schedule()
abstract fun reschedule()
fun enqueue(element: Runnable) {
addLast(element)
if (!scheduled) {
scheduled = true
schedule()
}
}
fun process() {
try {
// limit number of processed messages
repeat(yieldEvery) {
val element = removeFirstOrNull() ?: return@process
element.run()
}
} finally {
if (isEmpty) {
scheduled = false
} else {
reschedule()
}
}
}
}
// We need to reference global setTimeout and clearTimeout so that it works on Node.JS as opposed to
// using them via "window" (which only works in browser)
private external fun setTimeout(handler: dynamic, timeout: Int = definedExternally): Int
private external fun clearTimeout(handle: Int = definedExternally)
|
