Read Me First
The following was discovered as part of building this project:
- The JVM level was changed from '11' to '17', review the JDK Version Range on the wiki for more details.
Getting Started
Mono Stream Behaviour
Plain Mono Stream
Only one element in the stream, onNext is executed once, then the whole stream is completed.
@Test
public void MonoTest() {
Mono<?> stringMono = Mono
.just("Phil")
// .then(Mono.error(new RuntimeException("Intent to throw an exception")))
.log();
stringMono.subscribe(System.out::println, (error) -> System.out.println(error.getMessage()));
}
Output is as below
21:37:48.753 [Test worker] DEBUG reactor.util.Loggers -- Using Slf4j logging framework
21:37:48.759 [Test worker] INFO reactor.Mono.Just.1 -- | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
21:37:48.760 [Test worker] INFO reactor.Mono.Just.1 -- | request(unbounded)
21:37:48.760 [Test worker] INFO reactor.Mono.Just.1 -- | onNext(Phil)
Phil
21:37:48.760 [Test worker] INFO reactor.Mono.Just.1 -- | onComplete()
Mono Stream with error
Error interrupt the whole Mono stream, subscriber can't accept the only one element at all.
@Test
public void MonoTest() {
Mono<?> stringMono = Mono
.just("Phil")
.then(Mono.error(new RuntimeException("Intent to throw an exception")))
.log();
stringMono.subscribe(System.out::println, (error) -> System.out.println(error.getMessage()));
}
21:40:41.217 [Test worker] DEBUG reactor.util.Loggers -- Using Slf4j logging framework
21:40:41.223 [Test worker] INFO reactor.Mono.IgnoreThen.1 -- onSubscribe(MonoIgnoreThen.ThenIgnoreMain)
21:40:41.223 [Test worker] INFO reactor.Mono.IgnoreThen.1 -- request(unbounded)
21:40:41.224 [Test worker] ERROR reactor.Mono.IgnoreThen.1 -- onError(java.lang.RuntimeException: Intent to throw an exception)
21:40:41.224 [Test worker] ERROR reactor.Mono.IgnoreThen.1 --
java.lang.RuntimeException: Intent to throw an exception
at com.geekchow.webflux.MonoFluxTest.MonoTest(MonoFluxTest.java:12)
...
Intent to throw an exception
Flux Stream Behaviour
Plain Flux Stream
All elements within the Flux stream are accepted by subscriber one by one, and at last an onCompleted event occurs.
@Test
public void FluxTest() {
Flux<?> stringFlux = Flux
.just("Phil", "Zhou", "Full", "Stack", "Programmer")
.concatWithValues("AWS", "DevOps")
.log();
stringFlux.subscribe(System.out::println, (error) -> System.out.println(error.getMessage()));
}
Output of plain flux stream.
21:43:59.734 [Test worker] DEBUG reactor.util.Loggers -- Using Slf4j logging framework
21:43:59.741 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onSubscribe(FluxConcatArray.ConcatArraySubscriber)
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- request(unbounded)
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Phil)
Phil
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Zhou)
Zhou
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Full)
Full
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Stack)
Stack
21:43:59.742 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Programmer)
Programmer
21:43:59.743 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(AWS)
AWS
21:43:59.743 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(DevOps)
DevOps
21:43:59.743 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onComplete()
Flux stream with error element
Error element interrupt the Flux stream, but doesn't affect the elements before error.
@Test
public void FluxTest() {
Flux<?> stringFlux = Flux
.just("Phil", "Zhou", "Full", "Stack", "Programmer")
.concatWithValues("AWS", "DevOps")
.concatWith(Flux.error(new RuntimeException("Intent to throw an exception within Flux Stream")))
.concatWithValues("TypeScript", "Python")
.log();
stringFlux.subscribe(System.out::println, (error) -> System.out.println(error.getMessage()));
}
21:48:14.628 [Test worker] DEBUG reactor.util.Loggers -- Using Slf4j logging framework
21:48:14.635 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onSubscribe(FluxConcatArray.ConcatArraySubscriber)
21:48:14.636 [Test worker] INFO reactor.Flux.ConcatArray.1 -- request(unbounded)
21:48:14.636 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Phil)
Phil
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Zhou)
Zhou
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Full)
Full
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Stack)
Stack
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(Programmer)
Programmer
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(AWS)
AWS
21:48:14.637 [Test worker] INFO reactor.Flux.ConcatArray.1 -- onNext(DevOps)
DevOps
21:48:14.638 [Test worker] ERROR reactor.Flux.ConcatArray.1 -- onError(java.lang.RuntimeException: Intent to throw an exception within Flux Stream)
21:48:14.638 [Test worker] ERROR reactor.Flux.ConcatArray.1 --
java.lang.RuntimeException: Intent to throw an exception within Flux Stream
at com.geekchow.webflux.MonoFluxTest.FluxTest(MonoFluxTest.java:23)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
...
Intent to throw an exception within Flux Stream
Reference Documentation
For further reference, please consider the following sections:
- Official Gradle documentation
- Spring Boot Gradle Plugin Reference Guide
- Create an OCI image
- Spring Reactive Web
Guides
The following guides illustrate how to use some features concretely:
Additional Links
These additional references should also help you: