Reactor Java 1. MonoとFluxを作成します.

Mono And Flux are Lazy

Fluxを作成する一般的な方法

//1, 2, 3의 값을 가지는 Flux를 생성합니다.
Flux<Integer> integerFlux = Flux.just(1, 2, 3);

//"Hello", "foo", "bar"값을 가지는 Flux를 생성합니다.
Flux<String> stringFlux = Flux.just("Hello", "foo", "bar");

//Iterable한 요소로부터 Flux를 생성합니다. 여기서는 List가 사용되었습니다.
List<String> stringList = Arrays.asList("Hello", "foo", "bar");
Flux<String> fluxFromList = Flux.fromIterable(stringList);

//Java Stream에서도 동일하게 동작합니다.
Stream<String> stringStream = stringList.stream();
Flux<String> fluxFromStream = Flux.fromStream(stringStream);

//범위로 Flux 만들기
Flux<Integer> rangeFlux = Flux.range(1, 5); // -> Flux(1, 2, 3, 4, 5)

//100ms 마다 새로운 값을 만들어서 Flux를 생성한다. 값은 1부터 시작하여 증가한다.
Flux<Integer> intervalFlux = Flux.interval(Duration.ofMillis(100));

//다른 Flux또는 Mono로부터 생성
Flux<String> fluxCopy = Flux.from(fluxFromList);

Monoを作成する一般的な方法

//"Hello World!"를 가지는 Mono생성
Mono<String> helloWorld = Mono.just("Hello World");

//빈 Mono 생성
Mono<T> empty = Mono.empty();

//Callable로 부터 Mono 생성
Mono<String> helloWorldCallable = Mono.fromCallable(() -> "Hello World!");
Mono<User> user = Mono.fromCallable(UserService::fetchAnyUser);

//Future로 Mono 생성
CompletableFuture<String> helloWorldFuture = MyApi.getHelloWorldAsync();
Mono<String> monoFromFuture = Mono.fromFuture(helloWorldFuture);

//supplier로 부터 Mono생성
Ramdom rand = new Ramdom();
Mono<Double> MonoFromSupplier = mono.fromSupplier(rand::nextDouble);

//다른 Mono또는 Flux로 생성
Mono<Double> monoCopy = Mono.from(monoFromSupplier);
Mono<Integer> monoFromFlux = Mono.from(Flux.range(1, 10));

FluxまたはMonoを作成する一般的な方法

1) Error

2) Defer

//두 예제는 같은 결과를 가져다줍니다.
Integer getAnyInteger() throws Exception {
	throw new RuntimeException("An error as occured for no reason");
}


//두 메서드의 비교
Mono<Integer> fromCallable = Mono.fromCallable(this::getAnyInteger);
// result -> Mono.error(RuntimeException("An error as occured for no reason."))

Mono<Integer> defer = Mono.defer(() -> {
	try {
		Integer res = this.getAnyInteger();
		return Mono.just(res);
	} catch (Exception e) {
		return Mono.error(e);
	}
})
// result -> Mono.error(RuntimeException("An error as occured for no reason"))

3) Create

Integer getAnyInteger() throws Exception {
	throw new RuntimeException("An error as occured for no reason.");
}

Mono<Integer> = Mono.create(callback -> {
	try { callback.success(this.getAnyInteger()); }
	catch (Exception e) { callback.error(e); }
});

Flux<Double> flux = Flux.create(emitter -> {
	Random rnd = new Random();
	for(int i = 0; i <= 10; i++) emitter.next(rnd.nextDouble());

	int random = rnd.nextInt(2);

	if(random < 1) emitter.complete();
	else emitter.error(new RuntimeException("Bad luck, you had one change out of 2 to complete Flux));
});

Mono.just("Hello World.").subscribe(
	successValue -> System.out.println(successValue),
	error -> System.error.println(error.getMessage()),
	() -> System.out.println("Mono consumed.")
);

Hello World!
Mono consumed.

**the Error Message**
Mono consumed.

Flux.range(1, 5).subscribe(
	successValue -> System.out.println(successValue),
	error -> System.out.println(error.getMessage()),
	() -> System.out.println("Flux consumed.")
)

1
2
3
4
5
Flux comsumed.

1
2
3
// 이번에는 Flux comsumed. 가 출력되지 않는다.
// Flux는 에러가 발생하여 미래의 값을 다루는 것이 멈추면(모든 작업이 완전히 소비되지 않으면) Flux comsumed. 가 출력되지 않는다.