RxJava Osservabile ricevere una notifica sulla prima emissione

Question

ho tre osservabili che combino con combineLastest:

Observable<String> o1 = Observable.just("1"); 
    Observable<String> o2 = Observable.just("2"); 
    Observable<String> o3 = Observable.just("3"); 

    Observable.combineLatest(o1, o2, o3, new Func3<String, String, String, Object>() { 
     @Override 
     public Object call(String s, String s2, String s3) { 
      return null; 
     } 
    }); 

Voglio essere avvisato circa la prima emissione di uno degli osservabili senza ignorare le emissioni successive, che credo first operator farebbe. C'è un operatore conveniente per questo come (esempio):

o1.doOnFirst(new Func1<String, Void>() { 
     @Override 
     public Void call(String s) { 
      return null; 
     } 
    }) 

Answer 1

penso che si può avere un pratico doOnFirst con un semplice take se stai movimentazione un flusso:

public static <T> Observable<T> withDoOnFirst(Observable<T> source, Action1<T> action) { 
    return source.take(1).doOnNext(action).concatWith(source); 
} 

In questo modo l'azione è legata solo al primo elemento.

Questo potrebbe essere modificato per gestire osservabili che sono non appoggiate da flussi aggiungendo skip saltare gli articoli già adottate:

public static <T> Observable<T> withDoOnFirstNonStream(Observable<T> source, Action1<T> action) { 
    return source.take(1).doOnNext(action).concatWith(source.skip(1)); 
} 

Answer 2

Ci sono un paio di soluzioni che posso pensare. Il primo è un brutto ma semplice trucco di doOnNext. Basta aggiungere un campo booleano allo Action1 indicando se il primo elemento è stato ricevuto. Una volta ricevuto, fai quello che vuoi e gira il booleano. Per esempio:

Observable.just("1").doOnNext(new Action1<String>() { 

     boolean first = true; 

     @Override 
     public void call(String t) { 
      if (first) { 
       // Do soemthing 
       first = false; 
      } 
     } 
    }); 

La seconda è di sottoscrivere due volte sul osservabile si desidera monitorare utilizzando publish o share(), con una di quelle pubblicazioni che attraversano first (a seconda se si desidera collegare manualmente al pubblicato osservabile). Vi ritroverete con due osservabili separati che emettono gli stessi elementi, solo il primo si fermerà dopo la prima emissione:

ConnectableObservable<String> o1 = Observable.just("1").publish(); 

o1.first().subscribe(System.out::println); //Subscirbed only to the first item 
o1.subscribe(System.out::println); //Subscirbed to all items 

o1.connect(); //Connect both subscribers 

Answer 3

Uso rxjava-extras:

observable 
    .compose(Transformers.doOnFirst(System.out::println)) 

Si unità testato e sotto le copertine usano solo un contatore per abbonamento in un operatore. Si noti che per-subscription è importante in quanto vi sono molti casi di utilizzo in cui un'istanza osservabile viene utilizzata più di una volta e si desidera che l'operatore doOnFirst si applichi ogni volta.

Il codice sorgente è here.