I need also to integrate what we have in
observer_mock.go in the assertion API (ReactiveX/RxGo#109) and makes something more generic
if they did not follow the Github issue
I'm not a ReactiveX member so I don't known if people are going to listen to me but ok :)
my opinion about the future v2 of RxGo: https://github.com/ReactiveX/RxGo/issues/99#issuecomment-443816758
For the double subscription. In case of v2 second subscription would receive same elements of just sequence as the first?
here we just lose the messages as it is sent in a channel and retrieved during the first subscription
(and there's no way to retry it for the time being)
this function would get emitter as the argument and will be responsible for emitting sequence elements
Like empty - just emit complete
abd so forth
yet, for this very use-case I don't really get why we should have something different in RxGo
just my opinion obviously
we need something to manage a kind of publish/subscribe
either by implementing something from scratch, or reusing a proven library for such use case
has anyone an idea about something which already exists for this?
Hot observable can be implemented as follows: get create operator, pass closure with an emitter and subscription as arguments. If subscription is active (anything subscribed) use next, error etc. otherwise just pull an element and forget. Nothing is blocking. Hot source is working
I’m not sure about multicasting. Did you mean fan-out scenario or fan-in?
mmh, in your implementation, each call to next for example should be done in another goroutine in case where we have two subscriptions for example
otherwise the first observer would block the second one
you see what I mean?
Like in for e.g. in rxjava the reponsibility would be on connect and publish operator. I don’t remember the exact implementation but connect would buffer sequence elements and publish for each subscription create output channel. Then each time an element would appear it will be passed to all created output channels
I think that a specific operator could be a challenge like connect, publish pair or window. For me question is how to deal with error sequence. What is an error in golang? Imagine tha a right result of a function indicates error. What shall be emitted? Error with which argument? What about panic?
I don’t remember the exact implementation but connect would buffer sequence elements and publish for each subscription create output channel. Then each time an element would appear it will be passed to all created output channels
I understand. I believe, it's also time to think about back-pressure. Like having different strategies: buffer everything, drop messages not consumed on time etc.
Imagine, that at first you use create operator that wraps a channel. Without back pressure, each time something is on channel you call emitter.next with an element
With back pressure , first when there is request called on the observable constructed by create operator, you fetch up to requested number of elements to an intermediate buffer. In a emission loop you read from the buffer instead of wrapped channel
Let's say we have one hot observable and two observers
we may decide to publish each item in both channels