RxJava: create a self-dependent stream

Question

Say, I want to make an Observable in RxJava, which has a feedback coupling like on the image below.

I've managed to achieve that with the use of subjects, like this:

// Observable<Integer> source = Observable.range(0, 6); public Observable<Integer> getFeedbackSum(Observable<Integer> source) { UnicastSubject<Integer> feedback = UnicastSubject.create(); Observable<Integer> feedbackSum = Observable.zip(source, feedback.startWith(0), Pair::create) .map(pair -> pair.first + pair.second); feedbackSum.subscribe(feedback); return feedbackSum; }

It looks rather ugly. Is there a better way?

There is an operator for it: scan.

akarnokd
– akarnokd

2017-07-27 20:00:21 +00:00
Commented Jul 27, 2017 at 20:00 — akarnokd
– akarnokd, Commented Jul 27, 2017 at 20:00
Yeah, totally overlooked it. Thanks, @akarnokd!

4emodan
– 4emodan

2017-07-27 20:17:50 +00:00
Commented Jul 27, 2017 at 20:17 — 4emodan
– 4emodan, Commented Jul 27, 2017 at 20:17

akarnokd · Accepted Answer · 2017-07-28 18:03:32Z

There is an operator for this: scan:

public final <R> Observable<R> scan(R initialValue, BiFunction<R,? super T,R> accumulator)

Observable.range(0, 6) .scan(0, (a, b) -> a + b) .test() .assertResut(0, 1, 3, 6, 10, 15, 21);

In case your accumulator type is not immutable, you can use scanWith:

public final <R> Observable<R> scanWith(Callable<R> seedSupplier, BiFunction<R,? super T,R> accumulator)

scanWith example: static int initial = 1; .... Observable.range(1,2).scanWith(()->initial++, (a,b)->a+b).repeat(2).subscribe(out::println); . Result: 1,2,4,2,3,5

Collectives™ on Stack Overflow

RxJava: create a self-dependent stream

1 Answer 1

1 Comment

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Related