Closed iam closed 5 years ago
I like 'Change ref_count
to take another observable' the best.
avoid all the mutex overhead of publish/ref_count
To do this write a new subject that does no synchronization and use that with the multicast
operator to replace publish
. This only works if there is only one thread (or external synchronization) for both the producer and all calls to subscribe - also be careful of reentrancy.
Sounds good to me, I uploaded a PR with that approach you suggested works best.
When trying to design an algorithm that requires turning a regular rx chain into a diamond (imagine there's no multi-threading necessary here), there seems to be a lack of composability:
(A, B, C, D, E are some linear chains)
The (only working) approach I could figure out to this would look like:
The problem is that calling
connect
doesn't compose, so the whole graph can't just be expressed as anobservable<typeof(E)>
.What's the best practice to solve this? Should we change some operators to make this easier?
RxJava has
autoConnect
which rxcpp seems to lack.ref_count
would almost work except it callsconnect
too soon (whenB
is subscribed, but before waiting forC
).Possible solutions:
auto_connect
toconnectable_observable
(problem: it doesn't seem to call 'unsubscribe' ever inRxJava
. Not sure if this is acceptable?)ref_count
to take a threshold valueref_count(N)
: call connect() when reaching >=N subscribers. unsubscribe from source when reaching 0 subscribers (Today's behavior is simply N=1).ref_count
to take another observableref_count(Obs)
: subscribes/unsubscribes to its source as normal. also call Obs.connect() when reaching >=1 subscribers. (Today's behavior is simply Obs=source).publish
?fork(N) waits for N subscribers before subscribing to source. it forwards all the observer callbacks to all its subscribers. unsubscribe when subscriber count reaches 0.
(and this could also avoid all the mutex overhead of publish/ref_count).
What do you think? I'd be happy to write a PR if I knew what was the best way.
This has been bothering me for quite some time in some code I wrote for a project of mine, I ended up returning a
pair<observable<T>, connectable_observable<Y>>
the design of which "broke" immediately once I tried to integrate this into another larger observable chain.