RxJava2: Обратные вызовы PublishProcessor для первого подписанного и последнего незарегистрированного наблюдателя

Question

Я использую PublishProcessor<?> для обслуживания нескольких наблюдателей. Есть ли способ узнать, когда подписывается самый первый наблюдатель, и самый последний наблюдатель получает распоряжение?

Answer 1

Не напрямую, вам нужно захватить абонентов и подавители:

PublishProcessor<?> pp = ... 

AtomicInteger counter = new AtomicInteger(); 

Action onFirst = ... 

Action onLast = ... 

Flowable<?> f = pp.doOnSubscribe(s -> { 
    if (counter.getAndIncrement() == 0) { 
     onFirst.run(); 
    } 
}) 
.doFinally(() -> { 
    if (counter.decrementAndGet() == 0) { 
     onLast.run(); 
    } 
}) 

// use f for subscribe() instead of pp