RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Ok. I missed one detail. B can be merged with C
Observable<Long> a = Observable.interval(1, TimeUnit.SECONDS).map(x -> x + 1);
Observable<Long> b = a.map(v -> -v);
Observable<Long> c = Observable.interval(500, TimeUnit.MILLISECONDS).map(x -> x + 1000);
Observable.zip(a, Observable.merge(b, c), (v1, v2) -> "o1: " + v1 + ", o2: " + v2).subscribe(System.out::println);Then zip does wrong job too
Alternative, you could use debounce to filter out the double results when a and b both yield, at the risk of losing c when it yields too soon after b.
oh, and i just realize, you need something like
combineLatestRight. Does that exists in RxJava?
thanks a lot!
Maybe Scan?
Thats what i can come up with in 15 seconds ;)
I'm not 100% sure if this is possible, but i suppose it should be, but you can also use window or groupBy to combine all elements within 1 minute and then only take the first N elements from the resulting sequences. Doing that with live data is easy, but i'm sure it's possible with pre-existing timestamp data too.
It does depend on what state you had in mind
Subscriptions only receive events, they do not update the source.
PublishSubject<Integer> seq = PublishSubject.create();
Observable<Integer> seq2= seq.map(x ->{
timesCalled += 1;
return x * 2;
});
seq2.subscribe(e -> System.out.println("Hello 1: " + e));
seq2.subscribe(e -> System.out.println("Hello 2: " + e));
seq.onNext(1);
System.out.println("Times map called: " + timesCalled);map in this case called twice. (Is that a error or not?). And I get 2 different objects in each subscription, expected same.
If it helps to understand my problem
I assume timesCalled is declared outside the scope of map, that's where the problem lies.
Unless that was intended
Or publish seq2 so that side effects are shared between the observables instead.
It works. But as far as I understand doc,
publish creates ConnectableObservable. Which waits connect before emitting somethig. Sounds like not really aimed at current problem, right? Or I get it wrong? Or I need publish just to create new observable which keeps needed objects.
Until I
map it again by the way
specifically, is that a mean of propagating a backpressure request upstream?
say I have a bursty source, like
Observable.from(aCollectionOfThousandsOfStrings)
would that
flatMap signature with maxConcurrent of, say, 100 ensure that no more than 100 invocations of the map function are made at a time?
(effectively consuming 100 strings from the source, making the corresponding asynchronous calls, waiting for completion of at least one of them before makin a 101th call?)
The maxConcurrent refers to the max number of parallel active sources. It merges 100 collections and only start at #101 once one of them yields onComplete().
ok looks like it's what I'm after, a way of preventing a large list in an
Something like:
Observable.from(...) (or range) to swamp my io layer with too many requests in an instantSomething like:
Observable.range(1, 1000)
.map(id -> buildUrlToResource(id))
//driver can only cope with around 150 asynchronous requests
.flatMap(url -> driver.fireRequestAsync(url),
100)
.doOnNext(response -> showNotification(response.content())
.toBlocking().last(); //wait for last response
(I made a unit test in my project that demonstrates this behavior ;))
public class LookbackTransformer<T, R> implements Observable.Transformer<T, R> {
private final Func2<T, T, R> mergeFunc;
public LookbackTransformer(Func2<T, T, R> mergeFunc) {
this.mergeFunc = mergeFunc;
}
@Override
public Observable<R> call(Observable<T> source) {
Observable<T> offsetSource = source.skip(1);
return source.zipWith(offsetSource, mergeFunc);
}
}
for example it allows you to take a stream returned by
interval which emits [1, 2, 3, 4, 5] and map it to [(1,2),(2,3),(3,4),(4,5)]