I believe
like: we can use select with a default clause to implement non-blocking sends, receives, and even non-blocking multi-way selects.
I mean if we take the link, it actually prints "no message sent" if there is no receiver at the time where we try to send a message to a channel
mmh interesting, I was not familiar with this non-blocking send
thanks :)
either by publishing to a central channel and routing somehow to the observers
or by emitting elements directly to the observers
in RxJava it seems it's the second option: https://github.com/ReactiveX/RxJava/blob/0e7b8eaa61f9cac0538ef6a59bfbd0b119b87732/src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java#L171
basically if there is two subscriptions, they call the first
observer.onNext, wait for its completion then call the second observer.onNext
everything done synchronously
yep and we could easily implement strategies like drop for example by just checking if the target channel is available or not
I'm trying to implement a first design (based on the branch implementing real cold observables)
obs := Just(1, 2, 3).Publish()
got1 := make([]interface{}, 0)
obs.Subscribe(handlers.NextFunc(func(i interface{}) {
got1 = append(got1, i)
time.Sleep(200 * time.Millisecond) // Pause the observer on purpose
}), options.WithDropBackpressureStrategy())
obs.Connect()
time.Sleep(500 * time.Millisecond) // Ugly wait just for the example
fmt.Printf("%v\n", got1)obs := Just(1, 2, 3).Publish()
got1 := make([]interface{}, 0)
obs.Subscribe(handlers.NextFunc(func(i interface{}) {
got1 = append(got1, i)
time.Sleep(200 * time.Millisecond) // Pause the observer on purpose
}), options.WithBufferBackpressureStrategy(3))
obs.Connect()
time.Sleep(500 * time.Millisecond) // Ugly wait just for the example
fmt.Printf("%v\n", got1)
(branch
hot_observable)
in the first example, the observer is not able to consume all the items because of the
time.Sleep(200 * time.Millisecond) so it's just able to consume 1
They introduced specialized observable - Flowable. Thanks to flowable introduction, the back pressure is applied properly through whole pipe.
What do we have exactly in RxJava? Observable for cold observables, Flowable for hot observable with back-pressure and ConnectableObservable for hot observables without backpressure starting to publish once connect()is called?
this is really a first draft and any feedback is more than welcome
Hey! The v2 is progressing very well. Please check at the readme to get some insights on the main structural changes compared to the v2.
Any comment/remark is welcome!
Any comment/remark is welcome!
Hi everyone, I recently try to learn rxgo and I'm confused by some problems.
- Install rxgo with "go get -u github.com/reactivex/rxgo/v2" always report an error, because there is no v2 folder in the project. I have to use "go get -u github.com/reactivex/rxgo" to install.
- The example of "BufferWithTime" will panic, because of the "nil" options input.
- Do we have add/delete/update observer functions in rxgo? It seems these functions exist in rxjava. I want to delete/add observer to an observable, but I haven't found them.
