I'm using Reactive Spring for file upload and I can't quite understand how this works.
// Code 1
@PostMapping(value = "/upload", consumes = { MediaType.MULTIPART_FORM_DATA_VALUE })
public Mono<Void> upload(@RequestPart("key") Flux<Part> partFlux) {
return partFlux
.log(null, Level.INFO)
.doOnNext(part -> {
logger.info("[Return] " + part.name(), part.headers());
})
.then();
}
This receives a multipart request and logs its parts successfully.
However, the next version doesn't work:
// Code 2
@PostMapping(value = "/upload", consumes = { MediaType.MULTIPART_FORM_DATA_VALUE })
public void upload(@RequestPart("key") Flux<Part> partFlux) {
partFlux
.log(null, Level.INFO)
.doOnNext(part -> {
logger.info("[Subscribe] " + part.name(), part.headers());
})
.subscribe();
}
The difference is this does not returns but subscribes to the input Flux, which I expect to work as same as Code 1. However in the execution log, even though subscribe itself was successfully called, logging was not performed.
So I assumed that somehow Spring ignores/skips the inner subscription, but this was also not true.
@PostMapping(value = "/upload", consumes = { MediaType.MULTIPART_FORM_DATA_VALUE })
public void upload(@RequestPart("key") Flux<Part> partFlux) {
Flux.just(1, 2, 3, 4)
.log(null, Level.INFO)
.doOnNext(v -> {
logger.info("[Inner] " + v.toString());
})
.subscribe();
}
This successfully runs the inner subscription and creates log.
I'm confused why Code 1 and 2 works differently, mainly how doOnNext in Code 2 gets ignored.
Appendix
- Execution logs of Code 1
2021-10-18 14:28:41.932 TRACE 39084 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter : [c5c6915b-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:56565] HTTP POST "/upload", headers={masked}
2021-10-18 14:28:41.944 DEBUG 39084 --- [ctor-http-nio-2] s.w.r.r.m.a.RequestMappingHandlerMapping : [c5c6915b-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:56565] Mapped to com.healthhub.rxstudy.rest.FluxUploadController#upload(Flux)
2021-10-18 14:28:41.962 INFO 39084 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | onSubscribe([Fuseable] FluxFlattenIterable.FlattenIterableSubscriber)
2021-10-18 14:28:41.964 INFO 39084 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | request(unbounded)
2021-10-18 14:28:41.981 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : First boundary found @39 in PooledSlicedByteBuf(ridx: 0, widx: 228, cap: 228/228, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.982 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: PREAMBLE -> HEADERS
2021-10-18 14:28:41.982 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : End of headers found @47 in PooledSlicedByteBuf(ridx: 0, widx: 188, cap: 188/188, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.983 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Emitting headers: [Content-Disposition:"form-data; name="key""]
2021-10-18 14:28:41.986 TRACE 39084 --- [ctor-http-nio-2] o.s.http.codec.multipart.PartGenerator : Changed state: INITIAL -> FORM-FIELD
2021-10-18 14:28:41.986 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: HEADERS -> BODY
2021-10-18 14:28:41.987 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Boundary found @43 in PooledSlicedByteBuf(ridx: 0, widx: 140, cap: 140/140, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.987 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Emitting body: PooledSlicedByteBuf(ridx: 0, widx: 2, cap: 2/2, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.987 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: BODY -> HEADERS
2021-10-18 14:28:41.988 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : End of headers found @47 in PooledSlicedByteBuf(ridx: 0, widx: 96, cap: 96/96, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.988 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Emitting headers: [Content-Disposition:"form-data; name="key""]
2021-10-18 14:28:41.989 TRACE 39084 --- [ctor-http-nio-2] o.s.http.codec.multipart.PartGenerator : Emitting: DefaultFormFieldPart{key}
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.http.codec.multipart.PartGenerator : Changed state: FORM-FIELD -> FORM-FIELD
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: HEADERS -> BODY
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Boundary found @43 in PooledSlicedByteBuf(ridx: 0, widx: 48, cap: 48/48, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Emitting body: PooledSlicedByteBuf(ridx: 0, widx: 2, cap: 2/2, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: BODY -> HEADERS
2021-10-18 14:28:41.990 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Last boundary found in PooledSlicedByteBuf(ridx: 0, widx: 4, cap: 4/4, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1471, widx: 1471, cap: 2048))
2021-10-18 14:28:41.991 TRACE 39084 --- [ctor-http-nio-2] o.s.h.codec.multipart.MultipartParser : Changed state: HEADERS -> DISPOSED
2021-10-18 14:28:41.991 TRACE 39084 --- [ctor-http-nio-2] o.s.http.codec.multipart.PartGenerator : Emitting: DefaultFormFieldPart{key}
2021-10-18 14:28:41.991 TRACE 39084 --- [ctor-http-nio-2] o.s.h.c.m.MultipartHttpMessageReader : [c5c6915b-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:56565] Parsed parts [key] (content masked)
2021-10-18 14:28:41.992 INFO 39084 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | onNext(DefaultFormFieldPart{key})
2021-10-18 14:28:41.992 INFO 39084 --- [ctor-http-nio-2] c.h.rxstudy.rest.FluxUploadController : [Return] key
2021-10-18 14:28:41.992 INFO 39084 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | onNext(DefaultFormFieldPart{key})
2021-10-18 14:28:41.992 INFO 39084 --- [ctor-http-nio-2] c.h.rxstudy.rest.FluxUploadController : [Return] key
2021-10-18 14:28:41.992 INFO 39084 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | onComplete()
2021-10-18 14:28:41.993 TRACE 39084 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter : [c5c6915b-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:56565] Completed 200 OK, headers={masked}
2021-10-18 14:28:41.994 TRACE 39084 --- [ctor-http-nio-2] o.s.h.s.r.ReactorHttpHandlerAdapter : [c5c6915b-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:56565] Handling completed
- Execution logs of Code 2
2021-10-18 14:24:59.157 TRACE 37416 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter : [ff1c78e0-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:61525] HTTP POST "/upload", headers={masked}
2021-10-18 14:24:59.173 DEBUG 37416 --- [ctor-http-nio-2] s.w.r.r.m.a.RequestMappingHandlerMapping : [ff1c78e0-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:61525] Mapped to com.healthhub.rxstudy.rest.FluxUploadController#upload(Flux)
2021-10-18 14:24:59.190 INFO 37416 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | onSubscribe([Fuseable] FluxFlattenIterable.FlattenIterableSubscriber)
2021-10-18 14:24:59.191 INFO 37416 --- [ctor-http-nio-2] reactor.Flux.MonoFlattenIterable.1 : | request(unbounded)
2021-10-18 14:24:59.207 TRACE 37416 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter : [ff1c78e0-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:61525] Completed 200 OK, headers={masked}
2021-10-18 14:24:59.209 TRACE 37416 --- [ctor-http-nio-2] o.s.h.s.r.ReactorHttpHandlerAdapter : [ff1c78e0-1, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:61525] Handling completed