in rxJava auf dem gleichen Teilnehmer abonnieren 2

Question

In rxJava 1.x war es möglich, so etwas zu tun:

Observable.<Foo>create(subscriber -> { 
    ... 
    getNewObservable().subscribe(Subscribers.wrap(subscriber)); 
}) 

GetNewObservable() Rückkehr auch Observable<Foo>

ich nicht den gleichen Ansatz finden in rxJava 2. Es gibt keine Subscribe-Methode, die Emitter oder ObservableEmitter auf Observable-Objekt akzeptiert. Ich glaube, ich etwas tun kann:

Observable.<Foo>create((ObservableEmitter<Foo> emitter) -> { 
    ... 
    getNewObservable().subscribe(emitter::onNext, emitter::onError, emitter::onComplete); 
}) 

Aber ich frage mich, ob es eine einfachere Art und Weise wie in rxJava ist 1.

Dank.

Answer 1

1) Eine Möglichkeit, zu tun, was Sie in den Kommentaren beschrieben:

Observable.interval(...) 
    .map(n -> { 
      if (!permissionGranted) 
       throw new AccessDeniedException(); 
      ... // do work 
      return result; 
     }) 

2) Ein anderer Weg, es zu tun:

Observable.interval(...) 
    .flatMap(n -> { 
      if (permissionGranted) { 
       ... // do work 
       return Observable.just(result); 
      } else { 
       return Observable.error(new AccessDeniedException()); 
      } 
     }) 

3) Noch eine andere Art und Weise (überprüfen Erlaubnis nur einmal):

Observable.defer(() -> { 
      if (permissionGranted) { 
       return Observable.interval(...) 
        .map(n -> { 
         ... // do work 
         return result; 
        }); 
      } else { 
       return Observable.error(new AccessDeniedException()); 
      } 
     }) 

4) Eine weitere (die reaktive):

Completable.fromCallable(() -> { 
      if (!permissionGranted) 
       throw new AccessDeniedException(); 
      return true; // returned value does not matter 
     }) 
    .andThen(Observable.interval(...) 
        .map(n -> { 
         ... // do work 
         return result; 
        }) 
     ) 

, die Sie als Refactoring können:

askForPermission().andThen(getNewObservable())