Цитирование через наблюдаемый поток

Question

У меня есть поток наблюдаемых, который в основном эмулирует диаграмму состояния. Например:

whenThisHappens() 
    .switchMap(i -> { 
     if (i.isThisThing()) { 
      return whenThatHappens(); 
     } else { 
      return nothingHappened(); 
     } 
    } 
    .subscribe(thing -> { 
     if (thing.isThatThing()) { 
      log("Got that thing"); 
     } else { 
      log("Got nothing"); 
     } 
    }); 

Вопрос заключается в том, что я хочу, чтобы перебрать входа, пока какое-то событие не происходит (это в течение длительного работающей службы на Android). Сейчас я в состоянии сделать это, сохраняя наблюдаемым в переменной, подписавшись на него, а затем отписки и Повторная подписка на него в это onComplete

obs = whenThisHappens() 
    .switchMap(i -> { 
     if (i.isThisThing()) { 
      return whenThatHappens(); 
     } else { 
      return nothingHappened(); 
     } 
    } 
    .doOnNext(thing -> { 
     if (thing.isThatThing()) { 
      log("Got that thing"); 
     } else { 
      log("Got nothing"); 
     } 
    }) 
    .doOnComplete(i -> { 
     obs.unsubscribe(); 
     obs.subscribe(); 
    } 
    obs.subscribe(); 

Но я вроде чувствую, что я делаю что-то действительно неправильно Вот. Есть ли лучший способ сделать это? Я посмотрел на retry, но ошибки при отправке, чтобы повторить попытку, выглядят так же плохо, как и то, что я делаю сейчас.

Answer 1

Чтения кода это выглядит, как вы хотите filter:

whenThisHappens() 
     # ignore uninteresting things 
     .filter(i -> i.isThisThing()) 
     # do stuff on interesting things 
     .subscribe(item -> log("Got: " + item.toString())); 

Есть еще два дополнительных пылкие этот основной subscribe которые являются on-error функции и on-complete функции, которую вы можете использовать, если вам нужны - но подписки здесь управляется автоматом.

Answer 2

Я думаю, что вы пытаетесь сделать лучше, с PublishSubject или BehaviorSubject.

Поток будет публиковать предметы по этому вопросу, которые будут вызывать вашу подписку.

Вот поток событий класса я написал некоторое время назад:

public class SubjectEventStream implements IEventStream { 
    private final BehaviorSubject<IEvent> stream = BehaviorSubject.create(); 

    @Override 
    public void publish(Observable<IEvent> event) { 
     event.doOnNext(stream::onNext).subscribe(); 
    } 

    @Override 
    public Observable<IEvent> observe() { 
     return stream; 
    } 

    @Override 
    public <T> Observable<T> observe(Class<T> eventClass) { 
     return stream.ofType(eventClass); 
    } 
} 

Смотреть еще некоторую информацию здесь:

http://reactivex.io/documentation/subject.html

http://akarnokd.blogspot.com/2015/06/subjects-part-1.html