Zachowanie funkcji onNext i onComplete

Question

Mam narzędzie Observable, które robi coś bez potrzeby emitowania wartości. Mam też listę obiektów, które chcę, aby Observable działało. Więc dla wszystkich elementów tej listy: doSomething()

Observable.from(uris) 
     .flatMap(new Func1<Uri, Observable<Void>>() { 
      @Override 
      public Observable<Void> call(Uri uri) { 
       return createDoSomethingObservable(uri); 
      } 
     }) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribeOn(Schedulers.io()) 
     .subscribe(new Observer<Void>() { 
      @Override 
      public void onCompleted() { 
       Log.d(TAG, "completed"); 
      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onNext(Void aVoid) { 
       Log.d(TAG, "next"); 
      } 
     }); 

i metoda, która tworzy obserwowalnym:

Observable<Void> createDoSomethingObservable(final Uri uri) { 
    return Observable.create(new Observable.OnSubscribe<Void>() { 
     @Override 
     public void call(Subscriber<? super Void> subscriber) { 
      //doSomething 
      subscriber.onNext(null); 
      subscriber.onCompleted(); 
     } 
    }); 
} 

Teraz gdy uruchamiam to z listy z 3 elementów uzyskać:

next 
next 
next 
completed 

co jest dobre, ponieważ tego właśnie chciałem, ale nie wiem, dlaczego to działa. Najpierw zacząłem po prostu wzywać OnComplete, ponieważ w końcu obserwowalne wykonuje swoją pracę i wykonuje je. Ale wtedy oczywiście onNext nigdy nie jest wywoływany przez subskrybenta. To samo dotyczy odwrotności.

Więc moje pytania to:

1. Dlaczego onComplete tylko wezwał do ostatniego elementu listy?
2. Czy istnieje lepszy sposób rozwiązania tego problemu?

Answer 1

onComplete jest wywoływany jako ostatni element, ponieważ właśnie wtedy zakończył się najwcześniejszy możliwy do zaobserwowania w łańcuchu (from(uris)).

Oczekuje się, że Twoje obserwowalne emitowane przez flatMap będą wywoływać onComplete. Gdy to zrobisz (i powróciło call), można pracować nad kolejną emisją z from. Po zakończeniu emisji obserwowalnych from wywołuje onComplete i łańcuch jest zakończony, skutecznie.

Answer 2

onComplete (taki sam jak onError) jest wywoływana tylko raz w obserwowalnym łańcuchu, to jest sposób, który jest realizowany rxjava

myślę, że Twoje podejście jest poprawne, więc lepszy sposób nie jest potrzebne.

Answer 3

Myślę, że ten mały kod pomaga zrozumieć zachowanie onNext() i onComplete().
Załóżmy, że masz List<Uri>. Zmieńmy to ręcznie na Observable<Uri>ręcznie.

public static Observable<Uri> getUries(List<Uri> uriList){ 
    return Observable.create(new Observable.OnSubscribe<Uri>() { 
     @Override 
     public void call(Subscriber<? super Uri> subscriber) { 
      for(Uri uri : uriList){ 
       subscriber.onNext(uri); 
      } 

      subscriber.onCompleted(); 
     } 
    }); 
} 

lub przy użyciu lambda wyrażenia:

public static Observable<Uri> getUries(List<Uri> uriList){ 
    return Observable.create(subscriber -> { 
     for(Uri uri : uriList){ 
      subscriber.onNext(uri); 
     } 

     subscriber.onCompleted(); 
    }); 
} 

Jak widać, jesteśmy iteracji listy wejściowego i wywołać onNext() dla każdego elementu, a kiedy skończyliśmy transformujące naszych List do Observable nazwaliśmy onComplete()

PS Ten kod tylko demonstracja, proszę, nigdy go nie używaj do transfor List do Observable. Użyj operatora Observable.from().

UPDATE:

Operator from() realizacja:

... 
while (true) { 
    if (o.isUnsubscribed()) { 
     return; 
    } else if (it.hasNext()) { 
     o.onNext(it.next()); 
    } else if (!o.isUnsubscribed()) { 
     o.onCompleted(); 
     return; 
    } else { 
     // is unsubscribed 
     return; 
    } 
} 
... 

Link: https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java#L75-L87