RxJava - dziwne zachowanie .toList

Hej,
pobieram z API Forsquare listę wszystkich eventów w okolicy, korzystając z RxJavy. Generalnie jestem stosunkowo nowy w tym i nie do końca rozumiem co "tam się dzieje".

Zacznijmy od mojego interfejsu Retrofita:

@GET("/some/path")
Observable<Response> search(@QueryMap Map map); 

Wywołuję go tak:

     endpointService.search(requestMap)
            .onErrorReturn(throwable -> {
                L.e(TAG, "problems");
                return null;
            })
            .subscribe(rxBus::send);

Do tego korzystam z takiego tworu do eventów co się nazywa RxBus

    rxBus.register(Response.class)
            .subscribe(event -> {
                L.d(TAG, "Received search response");
            }); 

No i generalnie po każdym udanym puknięciu w serwer posyłam ten event do zainteresowanych fragmentów itp.
W moim przypadku odbieram go tak:

rxBus.register(Response.class)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .map(event -> event.getResponse().getList())
            .flatMap(Observable::from)
            .map(v -> new SearchItem(v.getName(), v.getAddress())
            .toList()
            .subscribe(event -> {
                L.w(TAG, "Stuff happens");
            }); 

No i szlag by to trafił, nie działa.
Co najlepsze przysiągłbym, że działało, ale w momencie implementacji adaptera itp. nagle przestało. Ech!
Co nie działa dokładnie? Ano - jeśli zakomentuję linikę toList() to ten log w subscribe mi się wywoła kilka czy kilkanaście razy. Nie mam pojęcia co robię źle.

Napisz może jaki masz problem, bo to że log ci się nie odpala "kilkanaście" razy to nie jest problem.
Dodatkowo jak wklejasz source code, to po to żebym mógł to odpalić, nie specjalnie stawiać takie środowisko żeby odpalić kawałek twojego kodu.
Napisz mały snipped żeby ktoś mógł wkleić i zobaczyć co nie gra.

Odnośnie pytania

jeśli zakomentuję linikę toList() to ten log w subscribe mi się wywoła kilka czy kilkanaście razy. Nie mam pojęcia co robię źle.

przecież toList

Returns an Observable that emits a single item, a list composed of all the items emitted by the source Observable.

Robiąc tu list, onNext() wykona się tylko raz, jakim prawem ma być parę razy "log" wypisany ?

Dodatkowo

po co ten toList ? jak czekasz na zakończenie to wbijasz się w onComplete, nie łapiesz błedu(onError) powinieneś,

.map(event -> event.getResponse().getList())
            .flatMap(Observable::from)

jest równe

.flatMap((event)->Obs.from(event.getResponse().getList()))

Przepraszam, może nie wysłowiłem się zbyt jasno:
mając .toList() nie wywołuje się NIC w .onSubscribe().

Nic, totalnie, na debugerze nie wchodzi nawet do tej metody. I nie wiem dlaczego.
Znaczy, są przesłanki, że przez busa nie przechodzi onCompleted, dlatego obiekty do listy są dodawane w nieskończoność... Nie wiem jednak czy jest to prawda a nawet jeśli, to jak to naprawić :/

Odpowiadając na Twoją edycję:
W odpowiedzi z serwera dostaję monstrualnego JSONa pełnego danych, których defacto nie chcę. Dlatego też chciałem tę listę jakoś "odchudzić". W pierwotnej wersji w mapie robiłem foreacha który tego Responsa zmieniał w List<Item>, którą chciałem prosto wpakować w adapter.
Generalnie tamto rozwiązanie chyba nadal by się sprawdzało, ale skoro już staram się być taki Rx, to próbuję rozwiązać to w sposób bardziej Rx.

dlatego obiekty do listy są dodawane w nieskończoność..

Podoba mi się Twoje rozumowanie :) witamy w rx'ach :D

na początek dodaj limit

 	limit(int num)

Nie wiem jednak czy jest to prawda a

wywal tego toList, i nadpisz onNext(w subsribie), powinien się cały czas wywoływać.
Ale to chyba bug, że stream jest nieskończony, szczególnie dla calla do REST API

A ja się tylko spytam co robi to to toList() ? i czy RxBus to jest z jakiejś biblioteki czy to jakiś twój Twór? bo z tego co rozumiem robisz sobie po prostu publishSubject do którego wypuszczasz eventy które się pojawiają po api callu.

To jeżeli chcesz tak zrobić to zrób se po prostu singletona który będzie robił api strzała, w singletonie będzie PublicSubject<Response> do którego będziesz dorzucał eventy.

A kazdy fragment czy co tam chcesz będziesz tylko subscribował na ten publishSubject w onCreateView i unsubscribe w onDestroyView żeby nie robić leaków.

albo pokaż co to jest to cudo rxBus