-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* switchMap
* debounce
* sample
* Update RxJava version to 2.2.8
Partially fixes #1107- Loading branch information
Showing
14 changed files
with
848 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| /* | ||
| * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
| */ | ||
|
|
||
| package kotlinx.coroutines | ||
|
|
||
| import kotlin.coroutines.* | ||
| import kotlin.jvm.* | ||
|
|
||
| private class VirtualTimeDispatcher(enclosingScope: CoroutineScope) : CoroutineDispatcher(), Delay { | ||
|
|
||
| private val originalDispatcher = enclosingScope.coroutineContext[ContinuationInterceptor] as CoroutineDispatcher | ||
| private val heap = ArrayList<TimedTask>() // TODO use MPP heap/ordered set implementation | ||
| private var currentTime = 0L | ||
|
|
||
| init { | ||
| /* | ||
| * Launch "event-loop-owning" task on start of the virtual time event loop. | ||
| * It ensures the progress of the enclosing event-loop and polls the timed queue | ||
| * when the enclosing event loop is empty, emulating virtual time. | ||
| */ | ||
| enclosingScope.launch { | ||
| while (true) { | ||
| val secret = ThreadLocalEventLoop.currentOrNull()?.processNextEvent() | ||
| ?: error("Event loop is missing, virtual time source works only as part of event loop") | ||
| if (secret <= 0) continue | ||
| if (secret > 0 && secret != Long.MAX_VALUE) error("Unexpected external delay: $secret") | ||
| val nextTask = heap.minBy { it.deadline } ?: return@launch | ||
| heap.remove(nextTask) | ||
| currentTime = nextTask.deadline | ||
| nextTask.run() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private inner class TimedTask( | ||
| private val runnable: Runnable, | ||
| @JvmField val deadline: Long | ||
| ) : DisposableHandle, Runnable by runnable { | ||
|
|
||
| override fun dispose() { | ||
| heap.remove(this) | ||
| } | ||
| } | ||
|
|
||
| override fun dispatch(context: CoroutineContext, block: Runnable) { | ||
| originalDispatcher.dispatch(context, block) | ||
| } | ||
|
|
||
| @ExperimentalCoroutinesApi | ||
| override fun isDispatchNeeded(context: CoroutineContext): Boolean = originalDispatcher.isDispatchNeeded(context) | ||
|
|
||
| override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle { | ||
| val task = TimedTask(block, currentTime + timeMillis) | ||
| heap += task | ||
| return task | ||
| } | ||
|
|
||
| override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) { | ||
| val task = TimedTask(Runnable { with(continuation) { resumeUndispatched(Unit) } }, currentTime + timeMillis) | ||
| heap += task | ||
| continuation.invokeOnCancellation { task.dispose() } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Runs a test ([TestBase.runTest]) with a virtual time source. | ||
| * This runner has the following constraints: | ||
| * 1) It works only in the event-loop environment and it is relying on it. | ||
| * None of the coroutines should be launched in any dispatcher different from a current | ||
| * 2) Regular tasks always dominate delayed ones. It means that | ||
| * `launch { while(true) yield() }` will block the progress of the delayed tasks | ||
| * 3) [TestBase.finish] should always be invoked. | ||
| * Given all the constraints into account, it is easy to mess up with a test and actually | ||
| * return from [withVirtualTime] before the test is executed completely. | ||
| * To decrease the probability of such error, additional `finish` constraint is added. | ||
| */ | ||
| public fun TestBase.withVirtualTime(block: suspend CoroutineScope.() -> Unit) = runTest { | ||
| withContext(Dispatchers.Unconfined) { | ||
| // Create a platform-independent event loop | ||
| val dispatcher = VirtualTimeDispatcher(this) | ||
| withContext(dispatcher) { block() } | ||
| ensureFinished() | ||
| } | ||
| } |
Oops, something went wrong.