Is there a way I can connect another callsite to a running c

Question

Is there a way I can connect another callsite to a running coroutine sort of like hot observable where observer can join in on the receiving of the value produced

gildor · Answer

You can use BroadcastChannel for this

ursus · Answer

What about if you have params for that cooroutine Youd need a key lt gt param mapping to get that Channel instance most likely this should be a returned from the function unless the caller chooses the id key but youll have different problems

ursus · Answer

and you cant just anonymously create that receiver where for observables this would be hidden

gildor · Answer

What do you mean about anonymous receiver

ursus · Answer

if the coroutine is parametrized in order to get those values youd need to get correct instance of channel

gildor · Answer

Hard to say with this description do you have some example maybe if you know how to do that example of Rx implementation

ursus · Answer

yes second

ursus · Answer

``` private val someObservablesMap = hashMap lt String Observable lt SomeResult gt gt fun someObservable someId String Observable lt SomeResult gt val obs = someObservable getOrPut someId Observable fromCallable or whatever return obs ```

ursus · Answer

callsite ``` someObservable abc subscribe ```

ursus · Answer

or even better if the params is not a ID just some data and those are mapped into an id internally

gildor · Answer

I don t see any connection here you create observable which will be started again on subscription

ursus · Answer

yes I omitted the hotness but okay ill add it in

ursus · Answer

``` class SomeManager private val someObservablesMap = hashMap lt String Observable lt SomeResult gt gt fun someObservable someId String Observable lt SomeResult gt val obs = someObservable getOrPut someId Observable fromCallable or whatever refCount return obs call site someObservable abc subscribe ```

gildor · Answer

With coroutine it s even easier if you need it one value use HashMap lt String suspend gt SomeResult gt

gildor · Answer

Exactly you not only omit that this coroutine is hot but also it returns single value so I really don t see why do you need Observable here

ursus · Answer

im not saying I need observable im researching coroutines and I know rx and this is what Im missing

gildor · Answer

It s hard to say what you want to do in this example but if you just want to cache result of async operation you can just use Deffered

ursus · Answer

because if you sideeffect to a Channel which isa different instance how would you tell the caller the id of the channel to which it needs to subscribe to

gildor · Answer

See to convert this snippet to coroutines you don t need channel

ursus · Answer

`HashMap lt String suspend gt SomeResult gt ` what would be that lamba I dont follow

gildor · Answer

Still not sure what you mean what is your use case Do you actually want to emit multiple events

gildor · Answer

This lambda is suspend function which replaces Observable fromCallable

ursus · Answer

No just one my issue is scoping say you have MessageManager postMessage postMessage needs to complete regardless of if UI which triggered it is still there

gildor · Answer

Because Observable fromCallable is essentially the same as suspend lambda

ursus · Answer

in Rx I d do this via hot observables and if UI comes back it reconnectes to the same observable in order to get the return value

gildor · Answer

But you cannot reconnect to cold Observable like in your example

ursus · Answer

my example is hot refCount I just forgot it sorry I added it in the edited sample

gildor · Answer

Hot observable in coroutines is Channel

gildor · Answer

For single value you don t need channel

ursus · Answer

well yes but not really since you need a way to grab the correct channel from the MessageManager its an indirection sideffecting of suspending function to the channel relay

gildor · Answer

Just use `async ` it will just cache result in Deffered

ursus · Answer

I dont think that will help since what if your UI instance goes away How would you connect to the channel once it comes back new UI instance

ursus · Answer

sec I ll write a sample

gildor · Answer

You just call openSubscription or asFlow or call await in case of Deferred

ursus · Answer

I think you meant this

ursus · Answer

``` suspend fun postMessage someText String val requestUuid = createUuiFromText someText channels getOrPut requestUuid Channel insert to database channels requestUuid send State SENDING apiClient actuallySend channels requestUuid send State SENT ```

ursus · Answer

my issue is how would you let the callsite ui know about the requestUuid key so it can grab the channel instance

ursus · Answer

you need some sort of a channel of channels

ursus · Answer

whereas with hot observables Id just subscribe to the observable with same param and Id get the same instance from the hashmap

ursus · Answer

basically the parametrization of the postMessage is the core issue you cannot just blindly MessageManager stateChannel

gildor · Answer

No i don t mean this

gildor · Answer

Quite similar implementation would be ``` private val someObservablesMap = hashMap lt String BroadcastChannel lt SomeResult gt gt fun someChannel someId String ReceiveChannel lt SomeResult gt val obs = someObservable getOrPut someId BroadcastChannel or whatever return obs openSubscription or asFlow someChannel abc consumeEach ```

gildor · Answer

This version don t have `refcount ` feature to unsubscribe only `publish ` to broadcast value

gildor · Answer

My main question is what `bservable fromCallable or whatever refCount ` should do

gildor · Answer

I mean you really want to unsubscribe if all subscribers are unsubscribed

ursus · Answer

ok Ill fill the body sec

gildor · Answer

If so there is no such operator as refCount in channel there is proposal to provide connectable Flow abstractions <https github com Kotlin kotlinx coroutines issues 1086>

ursus · Answer

yes im aware of flow im actually waiting for it since as I said flatmapping of singles is stupid regular suspending functions is better and flow as replacement for Observables and im good

ursus · Answer

here is the body

ursus · Answer

``` class MessageManager private val observablesMap = hashMap lt String Observable lt Unit gt gt fun postMessageObservable someText String Observable lt Unit gt val key = sha1 someText val obs = someObservable getOrPut key Observable fromCallable insert message to db with MessageState SENDING apiClient sendMessage update message to db with MessageState SENT publish autoConnect return obs ```

ursus · Answer

I know that Observable fromCallable is synchronous it could be flatMaps of singles but whatever doesnt matter for this example

gildor · Answer

`Observable fromCallable refCount ` is not valid code

gildor · Answer

You cannot refCount Observable only ConnectableObservable

ursus · Answer

yes sorry im writing from memory

gildor · Answer

Could you just explain semantics how this should work

ursus · Answer

I edited it added the publish changed it to autoconnect

ursus · Answer

okay so I want to post a message its a MessageManager responsibility and the posting action should be scoped to the MessageManager meaning it should only be canceled when MessageManager is closed and UI should just observe the result of this action i e show some toast Message sent or whatever

ursus · Answer

UI can come and go its android and postMessage should complete normalny regardless of if UI is there hence the publish autoConnect

ursus · Answer

but if you come back to the same screen again but different instance you should be able to hook back in to the running postMessage observable if its running and show the toast when finished

ursus · Answer

technically it should be Single not Obserable but publish is not defined on Single

gildor · Answer

gt publish is not defined on Single Exactly

ursus · Answer

but you can have observable tha only emits 1 and completes nothing wrong with that

ursus · Answer

ist just semantics

gildor · Answer

Isn t you need something like this ``` class MessageManager private val observablesMap = hashMapOf lt String Deferred lt Unit gt gt fun postMessageObservable someText String Deferred lt Unit gt val key = sha1 someText val obs = observablesMap getOrPut key GlobalScope async start = CoroutineStart LAZY insert message to db with MessageState SENDING apiClient sendMessage update message to db with MessageState SENT Unit return obs ```

gildor · Answer

~Not so idiomatic~ upd after fix more idiomatic that version with observable and you may have problems with not thread same hashmap but you already have it in your example

ursus · Answer

yes concurrent map but nevermind that

ursus · Answer

what would be the idiomatic way please

gildor · Answer

But semantically the same it s lazy will be shared with multiple consumers and forever cache value

ursus · Answer

suspend + sideffect to channel would be idiiomatic

gildor · Answer

sorry I just now realized that you don t need refCount feature here so you just start it once and cache forefere

gildor · Answer

no this is most idiomatic way imo

gildor · Answer

you even don t need Deferred

gildor · Answer

wait

ursus · Answer

its pseudo code yes youd clear the map in onFinally observablesMap remove this

gildor · Answer

Done ``` class MessageManager private val observablesMap = hashMapOf lt String Deferred lt Unit gt gt suspend fun postMessageObservable someText String val key = sha1 someText val obs = observablesMap getOrPut key GlobalScope async start = CoroutineStart LAZY insert message to db with MessageState SENDING apiClient sendMessage update message to db with MessageState SENT Unit return obs await ``` UPD Careful with scope GlobalScope here only for example

gildor · Answer

so no more streams just suspending function to post message

ursus · Answer

hmm nice btw that would multicast in case I needed it or cant I have multiple subs to it

gildor · Answer

there is no subscription

gildor · Answer

you just suspend

ursus · Answer

Id need a channel for that

gildor · Answer

you call it and receive result immidiately if message is delievered or suspend until it s sending

gildor · Answer

no

gildor · Answer

because Observable with 1 element is just a suspend function

ursus · Answer

I mean if multipleplaces need to know the result id need some pipe like Subject Channel I think

ursus · Answer

yes this is solved im following up

gildor · Answer

what I recommend to use in one of my first messages but semantics of you code was not clear for me

gildor · Answer

gt mean if multipleplaces need to know the result If you cache the same result there is no problem Deferred caches it

ursus · Answer

no no you solved it with the last messge im following up as how would you mutlicast this result to multiple places

ursus · Answer

Deffered is like a future

gildor · Answer

See <https kotlinlang slack com archives C1CFAFJSK p1560904898345800 thread ts=1560902795 338700 amp cid=C1CFAFJSK>

gildor · Answer

Yes Deffered is like a suspendable fututre

ursus · Answer

kk thanks a lot

gildor · Answer

But than you start using refCount which semantics is quite different it will not cache value forever only while someone is subscribed

ursus · Answer

btw do you have an opinion on the channel of channels

ursus · Answer

or observable lt observable lt PostMessageState gt gt

gildor · Answer

I would avoid it if I have better way but probably sometimes it s the only way

ursus · Answer

ursus · Answer

``` fun forceSync state offer SyncStatus SYNC GlobalScope launch val result = if load SyncStatus IDLE else SyncStatus FAILED state offer result ``` basically this

gildor · Answer

Also in my example there is GlobalScope it s just an example most probably you need own scope for MessageManager

ursus · Answer

but what if forceSync has some parameter I dont know like Target Foo Target Bar which it should sync

ursus · Answer

what would you do in this case please

gildor · Answer

What is Target Foo Anyway I would start new thread instead do not start new discussion here

ursus · Answer

ok I will