Блог

Главная - Вопросы по программированию - Пружинный поток doOnNext не выполнен

Пружинный поток doOnNext не выполнен

#spring #spring-webflux #project-reactor #multipart #flux

Вопрос:

Я использую реактивную пружину для загрузки файлов, и я не совсем понимаю, как это работает.

 // 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();
}

Это получает составной запрос и успешно регистрирует его части.

Однако следующая версия не работает:

 // 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();
}

Разница в том , что это не возвращает, а подписывается на ввод Flux , который, как я ожидаю, будет работать так же, как и код 1. Однако в журнале выполнения, даже несмотря subscribe на то, что он был успешно вызван, ведение журнала не было выполнено.

Поэтому я предположил, что каким-то образом Spring игнорирует/пропускает внутреннюю подписку, но это тоже было неправдой.

 @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();
}

Это успешно запускает внутреннюю подписку и создает журнал.

Я не понимаю, почему код 1 и 2 работают по-разному, в основном из-за того, что doOnNext в Коде 2 игнорируется.

Приложение

  • Журналы выполнения кода 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
  • Журналы выполнения кода 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

Комментарии:

1. Похоже, что параметр Flux<Часть>, переданный вашему контроллеру изначально, не связан с реальным анализатором разделов. Вероятно, они делают это только после вызова вашего метода контроллера. Если это так, они после вызова вашего метода подключают анализатор разделов к потоку, а затем подписываются на параметр Mono<Void>, который вы привязали к потоку в коде 1. В коде 2 вы пытаетесь подписаться на Flux до того, как он будет фактически привязан к анализатору разделов. Это всего лишь гипотеза, но она может объяснить наблюдаемое явление.

Метки: