RxJava Observable.concatMap metodu

Giriş

Açıklaması şöyle. Yani concatMap dışarıdaki Observable ile gelen girdinin sırasını değiştirmez. Ayrıca bir nesnenin işi bitmeden bir sonrakine geçmez.

The concatMap operator is actually quite similar to the flatMap, except that the operator waits for its inner publishers to complete before subscribing to the next one.

Örnek

Şöyle yaparız. Burada sıra korunuyor

@Test
public void flatMapVsConcatMap() throws Exception {
  System.out.println("******** Using flatMap() *********");
  Observable.range(1, 15)
    .flatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
    .subscribe(x -> System.out.print(x + " "));

  Thread.sleep(100);

  System.out.println("\n******** Using concatMap() *********");
  Observable.range(1, 15)
    .concatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
    .subscribe(x -> System.out.print(x + " "));

  Thread.sleep(100);
}

Çıktı şöyle

******** Using flatMap() *********

1 2 3 4 5 6 7 9 8 11 13 15 10 12 14

******** Using concatMap() *********

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

Örnek

Şöyle yaparız. Burada sıra korunuyor

@Test
void test_flatMap() {
  Flux.just(1, 2, 3)
    .flatMap(this::doSomethingAsync)
    //.flatMapSequential(this::doSomethingAsync)
    //.concatMap(this::doSomethingAsync)
    .doOnNext(n -> log.info("Done {}", n))
    .blockLast();
  }

private Mono<Integer> doSomethingAsync(Integer number) {
  //add some delay for the second item...
  return number == 2 ? 
    Mono.just(number).doOnNext(n -> log.info("Executing {}", n))
      .delayElement(Duration.ofSeconds(1)) :
    Mono.just(number).doOnNext(n -> log.info("Executing {}", n));
  }

Çıktı şöyle

// flatMap does not preserve original ordering, and has subscribed to all three elements 
// eagerly. Also, notice that element 3 has proceeded before element 2.
2022-04-22 19:38:49,164  INFO main - Executing 1
2022-04-22 19:38:49,168  INFO main - Done 1
2022-04-22 19:38:49,198  INFO main - Executing 2
2022-04-22 19:38:49,200  INFO main - Executing 3
2022-04-22 19:38:49,200  INFO main - Done 3
2022-04-22 19:38:50,200  INFO parallel-1 - Done 2

// flatMapSequential has subscribed to all three elements eagerly like flatMap 
// but preserves the order by queuing elements received out of order.
2022-04-22 19:53:40,229  INFO main - Executing 1
2022-04-22 19:53:40,232  INFO main - Done 1
2022-04-22 19:53:40,261  INFO main - Executing 2
2022-04-22 19:53:40,263  INFO main - Executing 3
2022-04-22 19:53:41,263  INFO parallel-1 - Done 2
2022-04-22 19:53:41,264  INFO parallel-1 - Done 3

//concatMap naturally preserves the same order as the source elements.
2022-04-22 19:59:31,817  INFO main - Executing 1
2022-04-22 19:59:31,820  INFO main - Done 1
2022-04-22 19:59:31,853  INFO main - Executing 2
2022-04-22 19:59:32,857  INFO parallel-1 - Done 2
2022-04-22 19:59:32,857  INFO parallel-1 - Executing 3
2022-04-22 19:59:32,857  INFO parallel-1 - Done 3