RxJava 2 Mise en route: exécuter des tâches en série. `andThen` ou` différer`?

Question

Je veux exécuter deux étapes dans RxJava en série. Je veux l'étape 1 à terminer avant l'étape 2 commence comme:

step 1: start 
step 1: finish 
step 2: start 
step 2: finish 

J'essaie différentes variations de l'API et RxJava exécute mes deux étapes en parallèle, ce qui est pas le comportement que je veux:

step 1: start 
step 2: start 
step 2: finish 
step 1: finish 

Dans l'exemple de code ci-dessous, j'essaie à la fois andThen et defer et j'obtiens une exécution parallèle. Comment puis-je résoudre ce problème afin qu'une seule étape s'exécute après la réussite de l'autre?

Le nom de la méthode andThen implique une exécution séquentielle en série. La méthode defer prend une fonction qui produit un autre Completable qui est la signature de méthode que je m'attendrais pour l'exécution de la tâche en série que je veux. Ne me donnez pas non plus le résultat que je veux.

Ai-je besoin de convertir en observable/fluide? Ou puis-je chaîner deux étapes avec Completable?

public class RxStep1Then2 { 
    public static Completable simulateCompletable(ScheduledExecutorService es, String msg, int msDelay) { 
     System.out.println(String.format("%s: start", msg)); 

     ScheduledFuture<?> future = es.schedule(() -> { 
      System.out.println(String.format("%s: finish", msg)); 
     }, msDelay, TimeUnit.MILLISECONDS); 

     return Completable.fromFuture(future); 
    } 

    public static void rxMain(ScheduledExecutorService es) { 
//  Completable c = simulateCompletable(es, "step 1", 1000) 
//    .andThen(simulateCompletable(es, "step 2", 500)); 

     Completable c = simulateCompletable(es, "step 1", 1000) 
       .defer(() -> simulateCompletable(es, "step 2", 500)); 

     c.blockingAwait(); 
     System.out.println("blockingAwait done"); 
    } 

    public static void main(String[] args) throws Exception { 
     ScheduledExecutorService es = Executors.newScheduledThreadPool(5); 
     System.out.println("Started ExecutorService."); 

     rxMain(es); 

     es.shutdown(); 
     es.awaitTermination(5, TimeUnit.MINUTES); 
     System.out.println("Shutdown ExecutorService. Done."); 
    } 
} 

Answer 1

L'exécution parallèle se produit parce que votre simulateCompletable commence les tâches avant que les Completable s sont même créés. Vous pouvez utiliser un Completable retardé directement:

Completable.fromAction(() -> System.out.println("First")) 
    .delay(1, TimeUnit.SECONDS) 
    .andThen(Completable.fromAction(() -> System.out.println("Second"))) 
    .blockingAwait(); 

Notez que

Completable c = simulateCompletable(es, "step 1", 1000) 
      .defer(() -> simulateCompletable(es, "step 2", 500)); 

ne chaîne aucune opération car defer est une méthode de fabrication statique qui crée une Completable indépendante; le premier simulateCompletableCompletable est simplement perdu.