ReactiveX/RxJava

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

Paco

@pakoito

with groupBy into sample you sort by type first, then group + latest in one operation

http://blogs.microsoft.co.il/iblogger/2015/08/11/animations-of-rx-operators-groupby/

yeah groupby doesn't require termination, it's just the marbles being silly

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

signalsForIds
  .buffer(1, SECONDS)
  .flatMap(listOfSignals -> Observable.from(listOfSignals).groupBy(signal -> signal.id).last())

Paco

@pakoito

fair enough

thanks

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

:+1:

Oh shi, no, it'll emit only last item from the last group

Paco

@pakoito

don't worry, change the buffer to a window

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

problem in flatMap

Paco

@pakoito

This message was deleted

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

The problem is to apply last() correctly so it 'll emit last item per group and not last item of the whole stream

Paco

@pakoito

This message was deleted

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

Again will emit last item of the last group :)

Paco

@pakoito

signalsForIds
  .groupBy(signal -> signal.id)
  .flatMap(obsSignals-> obsSignals.window(1, SECONDS).last())

I'll need to check on monday, np

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

Looks like an overhead because of using window per each group rather than window whole stream and then divide it to groups

Paco

@pakoito

it's more expensive, isn't it?

okay

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

yep, seems so. Anyway your problem is solvable, but requires some non-trivial combination of operators :smile: Write unit tests and experiment with implementation!

David Stemmer

@weefbellington

@pakoito just spitballing, but have you considered something like this?

Observable<Data> signal1 = hotSignal.filter(data -> data.id == 1).startWith(emptyData);
Observable<Data> signal2 = hotSignal.filter(data -> data.id == 2).startWith(emptyData);
Observable<Data> signal3 = hotSignal.filter(data -> data.id == 3).startWith(emptyData);

Observable<String> latestData = Observable.combineLatest(
    signal1, signal2, signal3,
    (data1, data2, data3) -> mergeData(data1, data2, data3)
);

latestData.sample(1, TimeUnit.SECONDS).subscribe(mergedData -> System.out.println(mergedData));

here is a gist to demonstrate: https://gist.github.com/weefbellington/952414cb7076976bfd49

David Stemmer

@weefbellington

the idea being -- you filter your main signal into multiple, one for each ID, and then sample the latest values from each filtered signal at timed intervals

Vasily Styagov

@VasyaFromRussia

Observable first = Observable.create(new Observable.OnSubscribe<Object>() {
                    @Override
                    public void call(Subscriber<? super Object> subscriber) {
                        subscriber.add(Subscriptions.create(() -> {
                            log(“first unsubscribe”);         
                        }));
                    }
                }).doOnUnsubscribe(() -> log(“first”));
   Observable second = Observable.create(…).doOnUnsubscribe(() -> log(“second”));
   Observable result = first.mergeWith(second).doOnUnsubscribe(() -> loge(“result”));
   Subscription subscription = result.subscribe(…);
   //…
   subscription.unsubscribe();

It logs only “result”. Looks like unsubscription is not propagated to merge’s child observables. So how to handle unsubscription inside of first observable’s Observable.OnSubscribe?

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

@weefbellington seems that your solution is suitable only for fixed/known variations of ids

David Stemmer

@weefbellington

@artem-zinnatullin there's no reason to couldn't use groupBy instead of filtee

David Stemmer

@weefbellington

errr that last thing was not something I meant to press enter on

but now I can't delete it

disregard the above comment

@artem-zinnatullin you're right, but if the number of streams is finite and the IDs are known, combineLatest is a pretty natural approach

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

yep, I agree, but usually I try to use more generic solutions just in case of changes in future :)

David Stemmer

@weefbellington

I'm playing around with an alternate solution using collect

David Stemmer

@weefbellington

@artem-zinnatullin @pakoito here is another solution that builds on the window/flatMap approach:

final Func0<HashMap<Integer, String>> initialState = () -> new HashMap<>();
final HashMap<Integer, String> output = new HashMap<>();

source.window(1, TimeUnit.SECONDS)
   .flatMap(observable -> observable.collect(
      initialState,
      (mostRecent, data) -> mostRecent.put(data.id, data.content))
   )
.map(input -> {
   output.putAll(input);
   return output;
}).subscribe(reduced -> System.out.println(output));

^ original gist updated: https://gist.github.com/weefbellington/952414cb7076976bfd49

that final .map is kind of kludgy, I'm pondering a better way to produce the output dictionary

Dorus

@Dorus

Isn't aggregate the operator that let you collect all element and forwards the result? Simply make a 1 second window and aggregate it into a map, since a map will only contain each key once, it will only keep the latest.

Dorus

@Dorus

Oh i found it, the correct operator here is called Collect, not Reduce. Both are variants of aggregate.

Dorus

@Dorus

nvm that's what you're doing

David Stemmer

@weefbellington

Yeah, collect is very similar to scan/reduce

Dorus

@Dorus

Also i'm not sure what's going on in the final map. You seem to add back all previous data (in case some id's where unused last second). Is that truly by the spec? If so, i would think there is no need to use window, just scan the original source instead of collect, and sample the result.

Also reduced == output, so you can just used reduced in subscribe.

David Stemmer

@weefbellington

@Dorus doesn't scan/reduce only emit a single value when the original source completes? In this case the source observable is hot, it may never complete. Maybe I'm misunderstanding what you're suggesting.

It doesn't look like reducedis the same as output to me, either. The output of each window is a Map. The final step is merging the map emitted by the current window with the map from the previous window.

Dorus

@Dorus

@weefbellington scan emit all intermediate values also. Not sure if that bode well if you are emitting the same mutable datatype each time.
The map ends with return output;. If i understand that correctly that means that in .subscribe(reduced -> ...) , reduced is the output variable returned by map.

David Stemmer

@weefbellington

@Dorus aha! So that is the difference between scan and reduce! I had thought they were aliases for the same operation. In that case, yes, scan with the values stored in a map is the way to go!

@Dorus thanks for pointing that out!

johnfoconnor

@johnfoconnor

is there a way to force an observable to emit it's result(s) "naturally" as a return value instead of as a callback within subscribe?

result = Observable.Range(1, 3).Sum().subscribeAndReturn()
print(result)  // 6

Artem Zinnatullin :slowpoke:

@artem-zinnatullin

no natural way, because it's a model of work — Observable pushes data to you.

You can block unit it finish via toBlocking

Where communities thrive

People

Repo info

Activity