jcornaz / miop Goto Github PK
View Code? Open in Web Editor NEWOperators library for kotlin coroutines channels
License: MIT License
Operators library for kotlin coroutines channels
License: MIT License
The folowing produce an exception:
emptyReceiveChannel<Int>().distinctUntilChanged().consumeEach { println(it) }
Exception in thread "main" kotlinx.coroutines.experimental.channels.ClosedReceiveChannelException: Channel was closed
at kotlinx.coroutines.experimental.channels.Closed.getReceiveException(AbstractChannel.kt:1009)
at kotlinx.coroutines.experimental.channels.AbstractChannel.receiveResult(AbstractChannel.kt:519)
at kotlinx.coroutines.experimental.channels.AbstractChannel.receive(AbstractChannel.kt:512)
at kotlinx.coroutines.experimental.channels.ChannelCoroutine.receive$suspendImpl(ChannelCoroutine.kt:31)
at kotlinx.coroutines.experimental.channels.ChannelCoroutine.receive(ChannelCoroutine.kt)
at com.github.jcornaz.miop.experimental.OperatorsKt$distinctUntilChanged$1.doResume(Operators.kt:156)
at com.github.jcornaz.miop.experimental.OperatorsKt$distinctUntilChanged$1.invoke(Operators.kt)
at com.github.jcornaz.miop.experimental.OperatorsKt$distinctUntilChanged$1.invoke(Operators.kt)
at com.github.jcornaz.miop.experimental.OperatorsKt$transform$1.doResume(Operators.kt:81)
at kotlin.coroutines.experimental.jvm.internal.CoroutineImpl.resume(CoroutineImpl.kt:42)
at kotlinx.coroutines.experimental.DispatchedKt.resumeCancellable(Dispatched.kt:209)
at kotlinx.coroutines.experimental.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:35)
at kotlinx.coroutines.experimental.CoroutineStart.invoke(CoroutineStart.kt:111)
at kotlinx.coroutines.experimental.AbstractCoroutine.start(AbstractCoroutine.kt:165)
at kotlinx.coroutines.experimental.channels.ProduceKt.produce(Produce.kt:95)
at kotlinx.coroutines.experimental.channels.ProduceKt.produce$default(Produce.kt:88)
at com.github.jcornaz.miop.experimental.OperatorsKt.transform(Operators.kt:81)
at com.github.jcornaz.miop.experimental.OperatorsKt.transform$default(Operators.kt:79)
at com.github.jcornaz.miop.experimental.OperatorsKt.distinctUntilChanged(Operators.kt:155)
It would be nice to be able to easily bind StateStore (#37) with JavaFx properties/collections
contract should be the same as for sequence in the standard library
It should be possible to bind subscribable variable to a subscribable value or a channel
If a channel emits collection, there could be an operator, computing the delta and emitting collection events.
The computation of the delta could then be made on a computation thread, and (for instance) ui could just apply events, in order to keep up to date its internal collections.
Unlike with async
and launch
which are meant launch background task, launchConsumeEach
is used by default to consume a channel, which means manly suspend until an element is ready. This is something we want to be on Unconfined
most of the time. So it should be the default.
Jitpack fails to publish version 0.1.0
because it uses Jdk 11.
Publish the artifacts on jcenter instead of relying on jitpack.
See:
fun <T> Iterable<T>.openSubscription(context: CoroutineContext = Unconfined, capacity: Int = 0): ReceiveChannel<T> =
asSequence().openSubscription(context, capacity)
fun <T> Sequence<T>.openSubscription(context: CoroutineContext = Unconfined, capacity: Int = 0): ReceiveChannel<T> =
produce(context, capacity = capacity) { forEach { send(it) } }
Complementary to #37, it should be easy to get a view of a StateStore
. A view would be another state store, with different state and action types, but delegating everything (subscription and dispatches) to the original store.
Example:
val originalStore: StateStore<S1, A1> = StateStore(intitialState)
val view: StateStore<S2, A2> = originalStore.map(
transformState = { it.subState },
transformAction = { adaptAction(it) },
)
This would allow to keep a single source of truth, while allowing to create some store dedicated to a task or aspect of the application and reduce the coupling for code which doesn't need to know the root state.
It is quite common to do:
subscribable.openSubscription().map { it.field }.distinctUntilChanged()
So it could be provided out-of the box as extension function over SubscribableValue
:
fun <T, R> SubscribableValue<T>.openSubscription(context: CoroutineContext = Unconfined, transform: suspend (T) -> R): ReceiveChannel<R> =
openSubscription().transform(context) { input, output ->
var previous = input.receive()
output.send(previous)
input.consumeEach { element ->
if (element !== previous) {
output.send(transform(it))
previous = element
}
}
}
This would mainly be useful for testing, But some use-case could be found for production code as well, allowing any custom operator to fails the reactive way.
Prodvide:
/**
* Start a job in the JavaFx thread which keeps up-to-date the [target] collection.
* Order of elements is ignored. Only consider the elements and their occurrence count.
*/
fun <E> ReceiveChannel<Collection<E>>.launchFxCollectionUpdater(target: MutableCollection<E>, parent: Job? = null): Job
Sometime dispatch is not enough and a function which would suspend until the event has been handled would be nice:
fun StateStore<S, E> : SubscribableValue<S> {
// other members here
/** suspend until the event has been handled and return the state resulting of the event */
suspend fun handle(event: E): S
}
suspend fun <T, R> ReceiveChannel<T>.transform(
context: CoroutineContext = Unconfined,
body: suspend (input: ReceiveChannel<T>, output: SendChannel<R>) -> Unit
) = produce<R>(context, onCompletion = consumes()) {
body(this@transform, channel)
}
As is the state store is meant to accept "events" describing the past and which need to be applied to the state in order to update it.
This is different from the concept of "action" (or "command") which describe an intent (from the user or the system) and which may produce events when executed.
state +---> +---------+
| exec | +---> event
action +---> +---------+
state +---> +---------+
| apply | +---> state
event +---> +---------+
This feature is a wild personal idea and deserve more time to get a refined API
Provide something similar to the redux store
This would be conceptually a sub-interface of SubscribableValue
.
Maybe something like that?
fun <A> StateStore<", A>.dispatch(initialAction: A, onError: (Throwable) -> A, finalAction: suspend () -> A) {
dispatch(initialAction)
launch(Unconfined) {
try {
dispatch(finalAction())
} catch(error: Throwable) {
dispatch(onError(error))
}
}
}
or:
interface AsyncAction<R, out A> {
/** pure */
val context: CoroutineContext get() = DefaultDispatcher
/** pure */
val onStart: A
/** pure */
fun onComplete(result: R): A
/** pure */
fun onError(error: Throwable): A
/**
* Do not have to be pure. May perform side effect and return different result over invocation.
*/
suspend fun execute(): R
}
fun <R, A> StateStore<*, A>.dispatch(action: AsyncAction<R, A>) {
dispatch(action.onStart)
launch(action.context) {
try {
dispatch(action.onComplete(action.execute()))
} catch (error: Throwable) {
dispatch(action.onError(error))
}
}
}
Use Executors.newCachedThreadPool
Implementation:
class Barrier private constructor(
private val lock: Mutex,
private val actualJob: Deferred<Unit> = async(Unconfined, start = CoroutineStart.LAZY) { lock.withLock { } }
) : Deferred<Unit> by actualJob {
constructor() : this(Mutex(true))
fun relase() {
lock.unlock()
}
}
Usage:
fun main(args: Array<String>) = runBlocking<Unit> {
val barrier = Barrier()
launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
barrier.await()
print("Hello ")
}
println("world")
barrier.release()
// barrier.cancel() can also be used to notify any suspended coroutines that the barrier will never be released.
}
kotlinx.coroutines version: 0.25.0-eap13
There should be one test class by operator. Some common behavior may be tested in a parent test class.
For example all operator should cancel the upstream channel if the result of the operator is cancelled.
kotlinx.coroutines provide an IO
dispatcher since 0.25.0
Disptachers.IO
for jvm modules.Use transform
(#36) and/or onCompletion
argument of produce
.
inline fun <reified R> ReceiveChannel<*>.filterIsInstance(): ReceiveChannel<R> = transform { input, output ->
input.consumeEach {
if (it is R) output.send(it)
}
}
Possible implementation:
fun <E> switchOnError(logError: (Throwable) -> Unit = {}, openChannel: () -> ReceiveChannel<E>): ReceiveChannel<E> = produce(Unconfined) {
while(isActive) {
try {
openChannel().consumeEach { send(it) }
} catch(t: Throwable) {
logError(t)
}
}
}
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.