RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
I'm not sure if that's exactly what you want
- subscriber will catch all items
- It needs to be deferred only if in
from(something)somethingshould be computed only after subscription to the observable
ah, sorry for indexes, markdown…
About
defer: imagine you have Observable.just(networkRequest())
To avoid that you need to have a function that wraps your call and
defer it:Observable
.defer(() -> Observable.just(networkRequest())
.map()
.subscribe()
this will be cold observable and
networkRequest() will be executed only after subscription.
I am doing an insert into DB wihich are single value inserts mostly coming from numerous clients, these are around 50k values coming in from different sources every 3 seconds, I am not able to do a batch insert into the db. Is their a way with obsevables that before calling on insert I can combine all of these into a single observable stream .. and all other observables are pushing into this defined observable ?
is there something that triggers an end-of-stream -- like a sentinal value that says "this is the last thing to be inserted into the database"?
or some sort of event that would allow you to figure out when the batch insert should occur?
here is buffer: http://reactivex.io/documentation/operators/buffer.html
looks like buffer supports sampling as well so that might be what you want
@eyeced ^
60k / sec
each time you need to push an event to the stream, you would call
PublishSubject#onNext
I could swear Sample only returns the latest value (if any) every time interval, while Buffer and Window allow you to group all events in a certain timespan together (Probably buffer in this case, as that returns the collected values in a collection, while window keeps them in a nested observable so you would still need to aggregate them). But i'm coming from Rx.NET, so it might be different here with RxJava.
I usually use sample to manage backpressure when I don't care about receiving every update
Also, combining streams is simply
Merge, as long as the clients dont emit errors. I see there is even a mergeDelayError in case you want to wait for all other clients to complete first. If bulk insert from a single client is easier than combining events from different clients, do the buffer operation before you merge.
Or connect to the clients inside the sequence with
flatMap.
If you work with the previous mentioned
PublishSubject<Observable<T>>, you can just call PublishSubject.OnNext(t)inside put. Do first subscribe to the PublishSubject with PublishSubject.asObservable().Merge().Subscribe(...) (doesn't strictly need the asObservable if you subscribe locally). Just make sure you don't expose the Subject anywhere.
I mean, i would love to call
Observer.Create(), but that doesn't allow you to call merge anymore. So you kinda need private Observer ob;
public constructor() {
PublishSubject<Observable<T>> sub = PublishSubject.create();
sub.merge().subscribe(...);
ob = sub.asObserver();
}
public void put(Observable<T> t) {
ob.onNext(t);
}
(Please anybody correct me if i'm wrong)