RxJava2 flatMap создает повторяющиеся события

Ваш PublishSubject подписан на несколько раз, по одному разу на элемент из interval().

Изменить: вам нужно будет каждый раз вводить новый PublishSubject (переключитесь на BehaviorSubject, если хотите сохранить первое/последнее излучение); передайте это длительному процессу и убедитесь, что его onComplete вызывается правильно, когда длительный процесс завершается.

Изменить

После недавних комментариев я мог бы придумать такое решение:

class MyBluetoothClient {
  private PublishSubject<BTLEEvent> statusPublishSubject = PublishSubject.create()

  public Observable<BTLEEvent> getEventObservable() {
    return statusPublishSubject
  }

  private void publishEvent(BTLEEvent event) {
    statusPublishSubject.onNext(event)
  }

  public void doStuff1() {
    // do something that returns:
    publishEvent(BTLEEvent.someEvent1)
  }

  public void doStuff2() {
    // do something else that eventually yields
    publishEvent(BTLEEvent.someEvent2)
  }
}

И вы используете его следующим образом:

MyBluetoothClient client = MyBluetoothClient()
client
  .getEventObservable()
  .subscribe( /* */ )

///

client.doStuff1()

/// 

client.doStuff2

Исходный ответ

Будет ли это делать?

Observable
  .interval(333, TimeUnit.MILLISECONDS)
  .flatMap(new Function<Long, ObservableSource<Integer>>() {
    private int val = 0;

    @Override public ObservableSource<Integer> apply(Long aLong) throws Exception {
      val++;
      return Observable.just(val);
      }
    })
  .subscribe(new Consumer<Integer>() {
    @Override public void accept(Integer integer) throws Exception {
      Log.w("value: %s", integer);
     }
  });

Итак, вот ответ, который я придумал. Я отмечу ответ @Tassos как правильный, поскольку он указал мне правильный путь.

Во-первых, мне нужен CachedSubject (субъект, который кэширует элементы, пока нет наблюдателей, и отправляет их, как только наблюдатель подключается), это необходимо, чтобы убедиться, что выбросы из apply действительно проходят. Класс в основном обертывает PublishSubject.

class CachedSubject<T> extends Subject<T> {

        private PublishSubject<T> publishSubject = PublishSubject.create();
        private Queue<T> cache = new ConcurrentLinkedQueue<>();

        @Override public boolean hasObservers() {
            return publishSubject.hasObservers();
        }

        @Override public boolean hasThrowable() {
            return publishSubject.hasThrowable();
        }

        @Override public boolean hasComplete() {
            return publishSubject.hasComplete();
        }

        @Override public Throwable getThrowable() {
            return publishSubject.getThrowable();
        }

        @Override protected void subscribeActual(Observer<? super T> observer) {
            while (cache.size() > 0) {
                observer.onNext(cache.remove());
            }
            publishSubject.subscribeActual(observer);
        }

        @Override public void onSubscribe(Disposable d) {
            publishSubject.onSubscribe(d);
        }

        @Override public void onNext(T t) {
            if (hasObservers()) {
                publishSubject.onNext(t);
            } else {
                cache.add(t);
            }
        }

        @Override public void onError(Throwable e) {
            publishSubject.onError(e);
        }

        @Override public void onComplete() {
            publishSubject.onComplete();
        }
    }

затем я использую этот класс с switchMap:

Observable
   .interval(1000, TimeUnit.MILLISECONDS)
   .switchMap(new Function<Long, ObservableSource<Integer>>() {

      private Subject<Integer> s = new CachedSubject<>();
      private int val = 0;

      @Override public ObservableSource<Integer> apply(Long aLong) throws Exception {
         val++;
         s.onNext(val);
         return s;
      }
   })
   .subscribe(new Consumer<Integer>() {
      @Override public void accept(Integer integer) throws Exception {
         Log.w("value: %s", integer);
      }
   });

Это фактически позволяет мне получать любое количество событий по методу apply<T t> и иметь только 1 Consumer, подписанный на него, получая от него все события.