[Spring WebFlux] 1. Mono와 Flux에 대한 이해(2) : 자주 사용하는 operator (method)

 

🪄 0. 이전 글

Mono와 Flux에 대한 이해는 이전 글에서 확인하실 수 있습니다.

[Spring WebFlux] 1. Mono와 Flux에 대한 이해(1) : 작동 방식과 이해

 

 

 

🏝 1. 새로운 시퀀스를 생성할 때 사용하는 Operator

상황에 따라 새로운 시퀀스를 생성할 때, 다양한 오퍼레이터를 만들 수 있습니다. 

 

1) T 타입의 객체를 방출하고, 해당 타입을 소지하고 있는 경우.

 

 

public static&nbsp;<T>&nbsp;Flux<T>&nbsp;just(T...&nbsp;data)

public static <T> Flux<T> just(T data);
public static <T> Mono<T> just(T data);

정적 팩토리 메서드로, Flux와 Mono는 just를 가지고 있습니다. 해당 T 타입의 객체를 소지하고 있는 경우이므로, just를 사용할 수 있습니다.

내가 직접적으로 여러개의 T 타입의 객체를 명시할 수 있다면, 열거하는 방식으로도 Flux를 만들 수 있습니다.

public static <T> Flux<T> just(T... data)

 

 

만약, null 혹은 Optional<T> 객체로부터 Mono 객체를 생성하고 싶다면, justOrEmpty 메서드를 사용하는 것이 좋습니다.

public static <T> Mono<T> justOrEmpty(@Nullable Optional<? extends T> data);
public static <T> Mono<T> justOrEmpty(@Nullable T data);

Parameter로, Optional 객체를 받아들이는 경우, Optional.isPresent라면 (비어있지 않다면) 해당 아이템을 방출 가능한 Mono를 생성합니다. 만약 Optional.isPresent가 아니라면, 오직 onComplete()를 호출하는 Mono를 생성하게 됩니다.

 

다른 Nullable한 객체의 경우, null이 아니라면 해당 아이템을 방출 가능한 Mono를, null 이면 오직 onComplete()를 호출하는 Mono를 생성하게 됩니다.

 

 

2) 메소드에 의해 리턴 받는 T 타입의 객체를 방출하는 경우.

 

 

이 역시 당연히도, 이전의 just 정적 팩토리 메서드를 사용할 수 있습니다.

public static <T> Flux<T> just(T... data);
public static <T> Flux<T> just(T data);
public static <T> Mono<T> just(T data);

 

하지만, lazily captured의 경우, 두 가지의 선택이 가능합니다.

※ lazily captured : subscribe하는 시점에 데이터를 생성하고 싶은 경우를 의미합니다.

1) Mono.fromSupplier의 사용
2) Flux.defer 혹은 Mono.defer의 내부를 Flux.just 혹은 Mono.just로 감싸기.

 

 

- Mono.just vs Mono.fromSupplier의 차이

public class MonoFromSupplier {

    public static void main(String[] args) {
        Mono<String> just1 = Mono.just(getName());
        Mono<String> just2 = Mono.fromSupplier(() -> getName());
    }
    private static String getName() {
        System.out.println("Generating name...");
        return Util.faker().name().fullName();
    }

}

 

위의 코드의 결과로는,

just1에서는 getName()이 실행되어 "Generating name..."이 실행되고, just2는 아무것도 출력되지 않습니다.

fromSupplier는 lazily captured로, subscribe하는 시점에 데이터를 생성하고, just2는 subscribe된 상태가 아니기 때문에 아무런 반응이 없던 것입니다.

 

 

 

public static <T> Mono<T> fromSupplier(Supplier<? extends T> supplier);

제공된 Supplier를 사용해 값을 만들어내는 Mono 객체를 만들어냅니다. 만약 Supplier가 Null임이 확인되면, 결과 Mono는 empty 상태로 완료합니다.

Mono의 fromSupplier는 supplier를 인자로 받기 때문에, 이 supplier를 누군가 invoke 해주어야 실행됩니다. 따라서 lazily captured가 가능합니다.

 

또 다른 방식으로는, Flux/Mono.defer 내부를 Flux/Mono.just로 감싸는 전략이 있습니다.

이 부분은 하단에서 "defer" operator에서 명시하겠습니다.

 

 

3) 반복되는 T 타입의 객체들이 주어진 경우

 

1) Array 즉, 배열이 주어진 경우
2) Collection 혹은 Iterable한 객체가 주어진 경우
3) 정수의 특정 범위를 사용하고 싶은 경우
4) 각 subscripction에 대해 제공되는 스트림을 사용하고 싶은 경우

 

3-1. Array 즉, 배열이 주어진 경우

public static <T> Flux<T> fromArray(T[] array);

주어진 배열에 포함된 아이템들을 방출하는 Flux를 생성하는 메소드는 fromArray입니다. 역시 정적 팩토리 메소드로 Flux 객체를 쉽게 생성할 수 있습니다.

 

 

3-2. Collection 혹은 Iterable한 객체가 주어진 경우

 

 

public static <T> Flux<T> fromIterable(Iterable<? extends T> it);

Iterable한 객체에 담겨있는 아이템을 방출하는 Flux 객체를 생성합니다. Iterable.iterator() 메서드는 적어도 한 번, 최대 두 번 각각의 subscriber를 호출할 것입니다.

 

fromIterable는 Iterable의 Spliterator를 검사해서, 반복이 유한하다는 것을 보장하는지를 판단합니다. 기본 Spliterator는 Iterable.iterator()을 두 번 호출할 수 있는 반복자를 감싸고 있으나, 이 두 번째 호출은 항상 유한한 것으로 간주되는 유형인 Collection에서 건너뛰게 됩니다.

 

 

3-3. 정수의 특정 범위를 사용하고 싶은 경우

 

public static Flux<Integer> range(int start, int count);

start 부터 차례로 늘어나는 정수 순열의 아이템을 방출하는 Flux를 만들어냅니다. [start, start+count)의 범위를 생성합니다.

 

 

3-4) 각 subscripction에 대해 제공되는 스트림을 사용하고 싶은 경우

public static <T> Flux<T> fromStream(Supplier<Stream<? extends T>> streamSupplier);

각각의 Subsricption을 위한 Supplier로부터 생성된 스트림의 아이템을 방출하는 Flux로 만들어냅니다. Stream은 오퍼레이터가 취소되거나, 에러상태이거나, 완료상태일 때 자동으로 닫힙니다.

 

 

4) 다양한 단일 값 소스로부터 방출되는 경우.

1) Supplier<T>
2) Task
3) CompletableFuture<T>

 

 

 

5) 그냥 완료하는 경우.

public static <T> Flux<T> empty();
public static <T> Mono<T> empty();

Mono와 Flux 모두 empty()라는 정적 스태틱 메서드를 가지고 있습니다. 완료하는 Mono/Flux 객체로 어떠한 아이템도 방출하지 않습니다. T 타입은 구체화된 Subscriber 타입입니다.

 

 

6) 에러를 발생시키는 경우.

public static <T> Flux<T> error(Throwable error);
public static <T> Mono<T> error(Throwable error);

public static <T> Flux<T> error(Supplier<? extends Throwable> errorSupplier);
public static <T> Mono<T> error(Supplier<? extends Throwable> errorSupplier);

 

 

7) 아무것도 하지 않는 경우.

public static <T> Flux<T> never();

어떠한 데이터, 혹은 에러 / 완료 신호도 보내지 않는 Flux를 생성합니다. Empty()는 Complete 신호를 보내는 것이 차이입니다.

Complete 신호도 보내지 않기 때문에 종료되지 않습니다.

 

도대체 이것을 어디에 쓰냐? 찾아본 결과 StackOverFlow의 답변 중 하나.

It is very often used in tests (typically to assert timeout behavior), but can also have production usage: some operators take a control Publisher as parameter in various situations where they need an asynchronous external signal to tell them to trigger some behavior. If in some cases you don't want said behavior to trigger, use never().

For instance, 'windowWhen' takes such parameter both for opening and closing windows (the later generated by a Function). Conditionally returning a Mono.never() you could have a window that never closes.

[출처] https://stackoverflow.com/questions/48273301/when-to-use-mono-never

결론은, 시간초과 행동을 단언하기 위한 테스트에서 종종 사용되며, 때때로는 프로덕션에서도 사용되기도 한다고 합니다. 몇몇 오퍼레이터는, 일부 동작을 트리거 하도록 하는 비동기 외부 신호가 필요한 다양한 상황에서 Publisher를 파라미터로써 제어합니다. 만약 해당 동작이 트리거되는 것을 원하지 않는다면 never를 사용합니다.

 

 

8) Subscribe할 때 결정되는 경우.

public static <T> Flux<T> defer(Supplier<? extends Publisher<T>> supplier);
public static <T> Mono<T> defer(Supplier<? extends Mono<? extends T>> supplier);

 

 

9) 일회용 자원에 의존하는 경우.

public static <T,D> Flux<T> using(Callable<? extends D> resourceSupplier,
                                  Function<? super D,? extends Publisher<? extends T>> sourceSupplier,
                                  Consumer<? super D> resourceCleanup)

public static <T,D> Mono<T> using(Callable<? extends D> resourceSupplier,
                                  Function<? super D,? extends Mono<? extends T>> sourceSupplier,
                                  Consumer<? super D> resourceCleanup)

 

 

10) 프로그래밍 방식으로 이벤트를 생성하는 경우. (상태를 사용할 수 있음)

 

 

 

🏝 2. 존재하는 시퀀스를 변환하기

 

..계속해서 추가 예정

 

 

 

🤷🏻‍♂️ 출처

[링크] https://guten-tag.tistory.com/481 (just vs fromSupplier)
[링크] https://projectreactor.io/docs/core/release/reference/index.html#which-operator (리액터 3 레퍼런스 가이드)
[링크] https://reactivex.io/documentation/operators/empty-never-throw.html (RX 레퍼런스)