RxJava 2: множественные наблюдатели не получают от одного наблюдаемого - PullRequest
Новая
       0

RxJava 2: множественные наблюдатели не получают от одного наблюдаемого

0 голосов
/ 19 января 2019

Я создаю наблюдаемую, на которую подписываются несколько наблюдателей. Но я вижу, что только один из наблюдателей на следующем звонке. Что я здесь не так делаю. 

        private Observable<String> getDataEmitter(String downloadFileName) {

           ObservableOnSubscribe<String> handler = emitter -> {
             String file = download(downloadFileName);
             if (file == null) {
               emitter.onError(e);
             }
             String[] tokens = file.split("\\n");
             for (int i = 0; i < tokens.length; i++) {
               emitter.onNext(token);
             }
             emitter.onComplete();
          };

          return Observable.create(handler);
        }

       public retrieve(String file) {

         final Observable<String> sourceObservable = getDataEmitter(file) 
          .flatMap(id -> {
            return Observable.from(service.find(id))
          }, Pair::of)
          .map(pair -> collect(request, pair));


         sourceObservable
           .flatMap(this::map)
           .map(this::fileFormat)
           .buffer(10)
           .subscribe(batched -> {
             System.out.println("b-1");
           }, err -> {
             System.out.println("error-1");
           },
           () -> {
             System.out.println("completed-1");
           });

        sourceObservable
          .map(pair -> format(pair))
          .subscribe(e -> {
            System.out.println("e-2" +e);
          },
          error -> System.out.println(error-2),
          () -> System.out.println("completed-2"));
        }

Вывод только из первой подписки. 

b-1
completed-1

почему не печатается вторая подписка? Я попытался использовать публикацию с подключением и даже воспроизвести. Но это не помогло. что я тут не так делаю?

1 Ответ

0 голосов
/ 19 января 2019

Ваша наблюдаемая завершена на .buffer(10), потому что buffers собирает 10 выбросов из наблюдаемой источника и испускает все это как пакетный объект, как только прекращается наблюдаемая источник ( см. Документацию ). Вы больше не будете наблюдать выбросы от него.

...