Как поделиться созданным Наблюдаемым?

Question

Я создал Наблюдаемый, который испускает ViewPager's positionOffset.

public Observable<Float> observable() { 

    return Observable.create(new Observable.OnSubscribe<Float>() { 

     @Override 
     public void call(final Subscriber<? super Float> subscriber) { 

      if (viewPager == null) { 
       if (!subscriber.isUnsubscribed()) { 
        subscriber.onError(new IllegalStateException("viewPager is null")); 
       } 
       return; 
      } 

      final ViewPager.OnPageChangeListener listener = new ViewPager.OnPageChangeListener() { 

       @Override 
       public void onPageScrolled(final int position, 
              final float positionOffset, 
              final int positionOffsetPixels) { 

        if (!subscriber.isUnsubscribed()) { 
         subscriber.onNext(positionOffset); 
        } 
       } 

       @Override 
       public void onPageSelected(final int position) { 

       } 

       @Override 
       public void onPageScrollStateChanged(final int state) { 

       } 
      }; 
      viewPager.addOnPageChangeListener(listener); 

      subscriber.add(Subscriptions.create(() -> viewPager.removeOnPageChangeListener(listener))); 
     } 
    }); 
} 

Работает до сих пор.

Но с каждой подпиской он создает новый ViewPager.OnPageChangeListener и добавляет его в ViewPager.

Есть ли способ, которым все подписки имеют одно и то же наблюдение, что ViewPager имеет только один прослушиватель?

Answer 1

Вы можете использовать Observable.share(), это хороший вариант, если у вас есть несколько подписчиков и не возражаете, если поздние подписчики могут потерять некоторые уведомления.

Если вы хотите, чтобы все подписчики видели одинаковые уведомления, используйте Observable.publish() - он вернет ConnectableObservable, который начинает испускать элементы после вызова метода connect(). Итак, вы можете сделать

connectable = originalObservable.publish(); 
connectable.subscribe(observer1); 
connectable.subscribe(observer2); 
connectable.connect(); 

Answer 2

Вы можете использовать Подключаемые наблюдаемые. Например приведенный ниже код имитирует бесконечный (никогда не завершает) поток событий:

Observable<Float> o = Observable.concat(Observable.just(1f,2f,3f), Observable.never()); 

Вы можете создать соединяемую наблюдаемую с помощью переигровки с RefCount, что будет кэшировать последнее значение:

Observable<Float> shared = o.replay(1).refCount(); 

Первым абонент сделает наблюдаемый «горячий» и заставит ее производить предметы:

shared.subscribe(new Action1<Float>() { 

    @Override 
    public void call(Float t) { 
    System.out.println("o1:" + t); 

    } 

}); 
Thread.sleep(1000); 

Второе наблюдаемое может подключиться к нему через некоторое время, и будет стартовать с последнего пункта, излучаемого:

shared.subscribe(new Action1<Float>() { 

    @Override 
    public void call(Float t) { 
    System.out.println("o2:" + t); 

    } 

}); 

Thread.sleep(1000); 

Выход:

o1:1.0 
o1:2.0 
o1:3.0 
o2:3.0