Rxjava Observe stream as List for Specific time

1
votes

What i am trying to do ?

I have an observable(dayListener) that emits 100 - 300 items(not confirmed maybe return nothing) from server first time I subscribe to it, then it returns 1 item after 1 hour. each Item is of type UserData

What is Problem?

I want to Observe item from (dayListner) for 3 seconds and convert it to List and store it , then later on keep on observing same observable for single item UserData

My Attempt

I cannot figure out how to convert observable for 3 seconds in rxjava 2 , because I can't return two types for same observable , UserData and List , so i am unable to form solution any ideas really appreciated !! , I'm just newbie to rxjava 2

2
androidarraylistlistenerrx-javarx-java2
Did you check the Buffer operator? – Pelocho
yes i did check that operator , but confusion is what about when i am observing single item after getting list of first emitted items , and buffer is risky operator it discards items sometimes – Zulqurnain Jutt
Well, you simply can't. An Observable emits a single type so you can't emit both a list and a simple item. You have some workarounds like emitting lists of a single item or emit something like an Either of a list or an item – Pelocho

2 Answers

3
votes

Simply use 2 different Observables, you're wanting to listen for 2 different types of data anyways.

First Observable should have a limit of 3 seconds:

firstListObservable = userSourceObservable .takeUntil(Observable.timer(3, TimeUnit.SECONDS)) .toList()

And then you have the Observable for the rest of the values:

remainingValuesObservable = userSourceObservable .skipUntil(Observable.timer(3, TimeUnit.SECONDS))

Was this what you wanted?

1
votes

You could create a Pair class and use Zip operator to merge both result in one emmited Item

class Pair {
        String a;
        Integer b;

        Pair(String a, Integer b) {
            this.a = a;
            this.b = b;
        }
    }

    @Test
    public void testZipDifferentTypes() {
        Observable.zip(obA(), obB(), Pair::new)
                .subscribe(System.out::println);
    }

    private Observable<String> obA() {
        return Observable.just("hello");
    }

    private Observable<Integer> obB() {
        return Observable.just(1);
    }