diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/PartEvent.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/PartEvent.java
index b8484f541e..3105a73687 100644
--- a/spring-web/src/main/java/org/springframework/http/codec/multipart/PartEvent.java
+++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/PartEvent.java
@@ -35,9 +35,9 @@ import org.springframework.util.Assert;
*
* Each part in a multipart HTTP message produces at least one
* {@code PartEvent} containing both {@link #headers() headers} and a
- * {@linkplain PartEvent#content() buffer} with content of the part.
+ * {@linkplain PartEvent#content() buffer} with the contents of the part.
*
- * - Form field will produce a single {@link FormPartEvent},
+ *
- Form fields will produce a single {@link FormPartEvent},
* containing the {@linkplain FormPartEvent#value() value} of the field.
* - File uploads will produce one or more {@link FilePartEvent}s,
* containing the {@linkplain FilePartEvent#filename() filename} used when
@@ -65,12 +65,12 @@ import org.springframework.util.Assert;
* // handle form field
* }
* else if (event instanceof FilePartEvent fileEvent) {
- * String filename filename = fileEvent.filename();
+ * String filename = fileEvent.filename();
* Flux<DataBuffer> contents = partEvents.map(PartEvent::content);
* // handle file upload
* }
* else {
- * return Mono.error("Unexpected event: " + event);
+ * return Mono.error(new RuntimeException("Unexpected event: " + event));
* }
* }
* else {
@@ -103,7 +103,7 @@ import org.springframework.util.Assert;
* .post()
* .uri("https://example.com")
* .body(Flux.concat(
- * FormEventPart.create("field", "field value"),
+ * FormPartEvent.create("field", "field value"),
* FilePartEvent.create("file", resource)
* ), PartEvent.class)
* .retrieve()
diff --git a/src/docs/asciidoc/web/webflux-functional.adoc b/src/docs/asciidoc/web/webflux-functional.adoc
index b403b92ac6..c34967fbb9 100644
--- a/src/docs/asciidoc/web/webflux-functional.adoc
+++ b/src/docs/asciidoc/web/webflux-functional.adoc
@@ -40,7 +40,7 @@ as the following example shows:
PersonRepository repository = ...
PersonHandler handler = new PersonHandler(repository);
- RouterFunction route = route()
+ RouterFunction route = route() <1>
.GET("/person/{id}", accept(APPLICATION_JSON), handler::getPerson)
.GET("/person", accept(APPLICATION_JSON), handler::listPeople)
.POST("/person", handler::createPerson)
@@ -64,6 +64,7 @@ as the following example shows:
}
}
----
+<1> Create router using `route()`.
[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"]
.Kotlin
@@ -202,21 +203,62 @@ Mono> map = request.multipartData();
val map = request.awaitMultipartData()
----
-The following example shows how to access multiparts, one at a time, in streaming fashion:
+The following example shows how to access multipart data, one at a time, in streaming fashion:
-[source,java,role="primary"]
+[source,java,indent=0,subs="verbatim,quotes",role="primary"]
.Java
----
-Flux parts = request.body(BodyExtractors.toParts());
+Flux allPartEvents = request.bodyToFlux(PartEvent.class);
+allPartsEvents.windowUntil(PartEvent::isLast)
+ .concatMap(p -> p.switchOnFirst((signal, partEvents) -> {
+ if (signal.hasValue()) {
+ PartEvent event = signal.get();
+ if (event instanceof FormPartEvent formEvent) {
+ String value = formEvent.value();
+ // handle form field
+ }
+ else if (event instanceof FilePartEvent fileEvent) {
+ String filename = fileEvent.filename();
+ Flux contents = partEvents.map(PartEvent::content);
+ // handle file upload
+ }
+ else {
+ return Mono.error(new RuntimeException("Unexpected event: " + event));
+ }
+ }
+ else {
+ return partEvents; // either complete or error signal
+ }
+ }));
----
-[source,kotlin,role="secondary"]
+
+[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"]
.Kotlin
----
-val parts = request.body(BodyExtractors.toParts()).asFlow()
+val parts = request.bodyToFlux()
+allPartsEvents.windowUntil(PartEvent::isLast)
+ .concatMap {
+ it.switchOnFirst { signal, partEvents ->
+ if (signal.hasValue()) {
+ val event = signal.get()
+ if (event is FormPartEvent) {
+ val value: String = event.value();
+ // handle form field
+ } else if (event is FilePartEvent) {
+ val filename: String = event.filename();
+ val contents: Flux = partEvents.map(PartEvent::content);
+ // handle file upload
+ } else {
+ return Mono.error(RuntimeException("Unexpected event: " + event));
+ }
+ } else {
+ return partEvents; // either complete or error signal
+ }
+ }
+ }
+}
----
-
-
[[webflux-fn-response]]
=== ServerResponse
diff --git a/src/docs/asciidoc/web/webflux-webclient.adoc b/src/docs/asciidoc/web/webflux-webclient.adoc
index 116480f985..e74d6f9d12 100644
--- a/src/docs/asciidoc/web/webflux-webclient.adoc
+++ b/src/docs/asciidoc/web/webflux-webclient.adoc
@@ -859,6 +859,54 @@ inline-style, through the built-in `BodyInserters`, as the following example sho
.awaitBody()
----
+==== `PartEvent`
+
+To stream multipart data sequentially, you can provide multipart content through `PartEvent`
+objects.
+
+- Form fields can be created via `FormPartEvent::create`.
+- File uploads can be created via `FilePartEvent::create`.
+
+You can concatenate the streams returned from methods via `Flux::concat`, and create a request for
+the `WebClient`.
+
+For instance, this sample will POST a multipart form containing a form field and a file.
+
+[source,java,indent=0,subs="verbatim,quotes",role="primary"]
+.Java
+----
+Resource resource = ...
+Mono result = webClient
+ .post()
+ .uri("https://example.com")
+ .body(Flux.concat(
+ FormPartEvent.create("field", "field value"),
+ FilePartEvent.create("file", resource)
+ ), PartEvent.class)
+ .retrieve()
+ .bodyToMono(String.class);
+----
+[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"]
+.Kotlin
+----
+var resource: Resource = ...
+var result: Mono = webClient
+ .post()
+ .uri("https://example.com")
+ .body(
+ Flux.concat(
+ FormPartEvent.create("field", "field value"),
+ FilePartEvent.create("file", resource)
+ )
+ )
+ .retrieve()
+ .bodyToMono()
+----
+
+On the server side, `PartEvent` objects that are received via `@RequestBody` or
+`ServerRequest::bodyToFlux(PartEvent.class)` can be relayed to another service
+via the `WebClient`.
+
[[webflux-client-filter]]
diff --git a/src/docs/asciidoc/web/webflux.adoc b/src/docs/asciidoc/web/webflux.adoc
index c57e012568..4f4aeb3e4c 100644
--- a/src/docs/asciidoc/web/webflux.adoc
+++ b/src/docs/asciidoc/web/webflux.adoc
@@ -619,11 +619,11 @@ https://github.com/synchronoss/nio-multipart[Synchronoss NIO Multipart] library.
Both are configured through the `ServerCodecConfigurer` bean
(see the <>).
-To parse multipart data in streaming fashion, you can use the `Flux` returned from an
-`HttpMessageReader` instead. For example, in an annotated controller, use of
-`@RequestPart` implies `Map`-like access to individual parts by name and, hence, requires
-parsing multipart data in full. By contrast, you can use `@RequestBody` to decode the
-content to `Flux` without collecting to a `MultiValueMap`.
+To parse multipart data in streaming fashion, you can use the `Flux` returned from the
+`PartEventHttpMessageReader` instead of using `@RequestPart`, as that implies `Map`-like access
+to individual parts by name and, hence, requires parsing multipart data in full.
+By contrast, you can use `@RequestBody` to decode the content to `Flux` without
+collecting to a `MultiValueMap`.
[[webflux-forwarded-headers]]
@@ -2825,29 +2825,98 @@ as the following example shows:
----
<1> Using `@RequestBody`.
+===== `PartEvent`
-To access multipart data sequentially, in streaming fashion, you can use `@RequestBody` with
-`Flux` (or `Flow` in Kotlin) instead, as the following example shows:
+To access multipart data sequentially, in a streaming fashion, you can use `@RequestBody` with
+`Flux` (or `Flow` in Kotlin).
+Each part in a multipart HTTP message will produce at
+least one `PartEvent` containing both headers and a buffer with the contents of the part.
+
+- Form fields will produce a *single* `FormPartEvent`, containing the value of the field.
+- File uploads will produce *one or more* `FilePartEvent` objects, containing the filename used
+when uploading. If the file is large enough to be split across multiple buffers, the first
+`FilePartEvent` will be followed by subsequent events.
+
+
+For example:
[source,java,indent=0,subs="verbatim,quotes",role="primary"]
.Java
----
- @PostMapping("/")
- public String handle(@RequestBody Flux parts) { <1>
- // ...
- }
+ @PostMapping("/")
+ public void handle(@RequestBody Flux allPartsEvents) { <1>
+ allPartsEvents.windowUntil(PartEvent::isLast) <2>
+ .concatMap(p -> p.switchOnFirst((signal, partEvents) -> { <3>
+ if (signal.hasValue()) {
+ PartEvent event = signal.get();
+ if (event instanceof FormPartEvent formEvent) { <4>
+ String value = formEvent.value();
+ // handle form field
+ }
+ else if (event instanceof FilePartEvent fileEvent) { <5>
+ String filename = fileEvent.filename();
+ Flux contents = partEvents.map(PartEvent::content);
+ // handle file upload
+ }
+ else {
+ return Mono.error(new RuntimeException("Unexpected event: " + event));
+ }
+ }
+ else {
+ return partEvents; // either complete or error signal
+ }
+ }));
+ }
----
<1> Using `@RequestBody`.
+<2> The final `PartEvent` for a particular part will have `isLast()` set to `true`, and can be
+followed by additional events belonging to subsequent parts.
+This makes the `isLast` property suitable as a predicate for the `Flux::windowUntil` operator, to
+split events from all parts into windows that each belong to a single part.
+<3> The `Flux::switchOnFirst` operator allows you to see whether you are handling a form field or
+file upload.
+<4> Handling the form field.
+<5> Handling the file upload.
[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"]
.Kotlin
----
@PostMapping("/")
- fun handle(@RequestBody parts: Flow): String { // <1>
- // ...
- }
+ fun handle(@RequestBody allPartsEvents: Flux) = { // <1>
+ allPartsEvents.windowUntil(PartEvent::isLast) <2>
+ .concatMap {
+ it.switchOnFirst { signal, partEvents -> <3>
+ if (signal.hasValue()) {
+ val event = signal.get()
+ if (event is FormPartEvent) { <4>
+ val value: String = event.value();
+ // handle form field
+ } else if (event is FilePartEvent) { <5>
+ val filename: String = event.filename();
+ val contents: Flux = partEvents.map(PartEvent::content);
+ // handle file upload
+ } else {
+ return Mono.error(RuntimeException("Unexpected event: " + event));
+ }
+ } else {
+ return partEvents; // either complete or error signal
+ }
+ }
+ }
+}
----
<1> Using `@RequestBody`.
+<2> The final `PartEvent` for a particular part will have `isLast()` set to `true`, and can be
+followed by additional events belonging to subsequent parts.
+This makes the `isLast` property suitable as a predicate for the `Flux::windowUntil` operator, to
+split events from all parts into windows that each belong to a single part.
+<3> The `Flux::switchOnFirst` operator allows you to see whether you are handling a form field or
+file upload.
+<4> Handling the form field.
+<5> Handling the file upload.
+
+Received part events can also be relayed to another service by using the `WebClient`.
+See <>.
[[webflux-ann-requestbody]]