KindConnection #1199
KindConnection #1199
Conversation
There was a problem hiding this comment.
Looks good overall but feels like the connection arguments required for the async() calls are being ignored and each data type instantiates its own connection inside the method body and the param conn is ignored. Should we keep it exposed and provide instances as default values? or should we remove the connection from the Proc aliases?
| @@ -15,6 +15,8 @@ interface Applicative<F> : Functor<F> { | |||
|
|
|||
| fun <A> A.just(dummy: Unit = Unit): Kind<F, A> = just(this) | |||
|
|
|||
| fun unit(): Kind<F, Unit> = just(Unit) | |||
| @@ -87,7 +87,7 @@ interface DeferredKMonadDeferInstance : MonadDefer<ForDeferredK>, DeferredKBrack | |||
| @extension | |||
| interface DeferredKAsyncInstance : Async<ForDeferredK>, DeferredKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): DeferredK<A> = | |||
| DeferredK.async(fa = fa) | |||
| DeferredK.async(fa = { _, cb -> fa(cb) }) | |||
There was a problem hiding this comment.
don't we need to pass in the connection from the callers? looks like we always ask for it in the async arguments (DeferredKProc) but then it's being instantiated inside the method itself and the param ignored?
There was a problem hiding this comment.
This is just for now.
This PR just adds the support on the concrete types. This is just in an attempt to do everything in smaller PRs. A discussion of where goes what needs to follow still, I think me & Paco are a little out of sync on this.
There was a problem hiding this comment.
This seems like the correct approach, yes. The correct implementation (the one in Concurrent) has to have a connection parameter, and the Async one is an specialization where the parameter is ignored.
| @@ -115,7 +115,7 @@ interface FluxKAsyncInstance : | |||
| Async<ForFluxK>, | |||
| FluxKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): FluxK<A> = | |||
| FluxK.runAsync(fa) | |||
| FluxK.runAsync { _, cb -> fa(cb) } | |||
There was a problem hiding this comment.
same here, what's the intention on exposing it to callers if it's ignored inside ?
| @@ -88,7 +88,7 @@ interface MonoKAsyncInstance : | |||
| Async<ForMonoK>, | |||
| MonoKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): MonoK<A> = | |||
| MonoK.async(fa) | |||
| MonoK.async { _, cb -> fa(cb) } | |||
| @@ -119,7 +119,7 @@ interface FlowableKAsyncInstance : | |||
| Async<ForFlowableK>, | |||
| FlowableKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): FlowableK<A> = | |||
| FlowableK.async(fa, BS()) | |||
| FlowableK.async({ _, cb -> fa(cb) }, BS()) | |||
| @@ -107,7 +107,7 @@ interface MaybeKMonadDeferInstance : MonadDefer<ForMaybeK>, MaybeKBracketInstanc | |||
| @extension | |||
| interface MaybeKAsyncInstance : Async<ForMaybeK>, MaybeKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): MaybeK<A> = | |||
| MaybeK.async(fa) | |||
| MaybeK.async { _, cb -> fa(cb) } | |||
| @@ -86,7 +86,7 @@ interface SingleKAsyncInstance : | |||
| Async<ForSingleK>, | |||
| SingleKMonadDeferInstance { | |||
| override fun <A> async(fa: Proc<A>): SingleK<A> = | |||
| SingleK.async(fa) | |||
| SingleK.async { _, cb -> fa(cb) } | |||
| import arrow.typeclasses.Applicative | ||
| import java.util.concurrent.atomic.AtomicReference | ||
|
|
||
| typealias CancelToken<F> = Kind<F, Unit> |
...cts/arrow-effects-kotlinx-coroutines-instances/src/test/kotlin/arrow/effects/DeferredTest.kt
Outdated
Show resolved
Hide resolved
| * val result = DeferredK.async { conn: DeferredKConnection, cb: (Either<Throwable, String>) -> Unit -> | ||
| * val resource = Resource() | ||
| * conn.push(DeferredK { resource.close() }) | ||
| * resource.asyncRead { value -> cb(value.right()) } |
There was a problem hiding this comment.
This is a probable cause for the explosion in CI. Ank can't handle asynchrony of any kind.
| * | ||
| * Matching the behavior of [async], | ||
| * Matching the behavior of [asyncK], | ||
| * its [CoroutineContext] is set to [DefaultDispatcher] | ||
| * and its [CoroutineStart] is [CoroutineStart.LAZY]. | ||
| * | ||
| * {: data-executable='true'} | ||
| * | ||
| * ```kotlin:ank |
There was a problem hiding this comment.
Do we need Ank for this when it has no return?
modules/effects/arrow-effects-kotlinx-coroutines/src/main/kotlin/arrow/effects/DeferredK.kt
Outdated
Show resolved
Hide resolved
| CompletableDeferred<A>().apply { | ||
| fa { | ||
| fa(conn) { |
modules/effects/arrow-effects-kotlinx-coroutines/src/main/kotlin/arrow/effects/DeferredK.kt
Outdated
Show resolved
Hide resolved
|
I've discussed with Simon directly the next step to take. The current implementations of That means we can have the cancel silently or cancel throwing semantics that IO has, for everyone :D |
modules/effects/arrow-effects-rx2-instances/src/test/kotlin/arrow/effects/ObservableKTests.kt
Outdated
Show resolved
Hide resolved
| @@ -503,7 +503,9 @@ sealed class DeferredK<A>( | |||
| fun <A> async(scope: CoroutineScope = GlobalScope, ctx: CoroutineContext = Dispatchers.Default, start: CoroutineStart = CoroutineStart.LAZY, fa: DeferredKProc<A>): DeferredK<A> { | |||
| val conn = DeferredKConnection() | |||
| return Generated(ctx, start, scope) { | |||
| CompletableDeferred<A>().apply { | |||
| val supervisor = Job() | |||
There was a problem hiding this comment.
does the parent of this job need to be within the scope? See if you can create it from there.
| @@ -502,7 +502,8 @@ sealed class DeferredK<A>( | |||
| */ | |||
| fun <A> async(scope: CoroutineScope = GlobalScope, ctx: CoroutineContext = Dispatchers.Default, start: CoroutineStart = CoroutineStart.LAZY, fa: DeferredKProc<A>): DeferredK<A> { | |||
| val conn = DeferredKConnection() | |||
| val supervisor = Job() | |||
| //If the context doesn’t have a Job, then the coroutine which is created doesn’t have a parent. | |||
| val supervisor = scope.coroutineContext[Job] ?: Job() | |||
There was a problem hiding this comment.
In scope.coroutineContext[Job], what is Job here? where is it defined? I'd like to know to understand the architecture better.
| @@ -502,7 +502,8 @@ sealed class DeferredK<A>( | |||
| */ | |||
| fun <A> async(scope: CoroutineScope = GlobalScope, ctx: CoroutineContext = Dispatchers.Default, start: CoroutineStart = CoroutineStart.LAZY, fa: DeferredKProc<A>): DeferredK<A> { | |||
| val conn = DeferredKConnection() | |||
| val supervisor = Job() | |||
| //If the context doesn’t have a Job, then the coroutine which is created doesn’t have a parent. | |||
| val supervisor = scope.coroutineContext[Job] ?: Job() | |||
There was a problem hiding this comment.
Do we need to create a Job() for a GlobalScope?
This PR replaces
IOConnectionwith an abstract versionKindConnectionwhich is wired for allFthat currently supportAsync#async.