WebFlux multipart streaming not work

[DVMaslov]

I want to upload files without saving to disk or memory.

Controller:

@RestController
@RequestMapping("api/v1/files")
public class FilesController {

    @PostMapping
    public Mono<String> upload(@RequestPart FilePart filePart) throws IOException {
        filePart.transferTo(Files.createTempFile("test", filePart.filename()));
        return Mono.just(filePart.filename());
    }
}

Config:

@Configuration
public class WebConfig implements WebFluxConfigurer {

    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {
        DefaultPartHttpMessageReader partReader = new DefaultPartHttpMessageReader();
        partReader.setStreaming(true);
        configurer.defaultCodecs().multipartReader(new MultipartHttpMessageReader(partReader));
    }
}

Request:

curl --location --request POST 'http://localhost:8080/api/v1/files' --form 'file=@"/some_file.pdf"'

Full example:
streaming-multipart.zip

[poutsma]

The streaming mode of the DefaultPartHttpMessageReader works, but is a bit rough around the edges.

In order to make it work, I had to change the WebConfig to

    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {
        DefaultPartHttpMessageReader partReader = new DefaultPartHttpMessageReader();
        partReader.setStreaming(true);
        configurer.customCodecs()
                .register(partReader);
    }

I had to change the controller to

@RestController
@RequestMapping("api/v1/files")
public class FilesController {

    @PostMapping
    public Flux<String> upload(@RequestBody Flux<Part> parts) {
        return parts.flatMap(part -> {
            if (part instanceof FilePart) {
                FilePart filePart = (FilePart) part;
                String filename = filePart.filename();
                return createTempFile("test", filename)
                        .flatMap(filePart::transferTo)
                        .then(Mono.just(filename));
            } else {
                return part.content().map(DataBufferUtils::release)
                        .then(Mono.empty());
            }
        });
    }

    private static Mono<Path> createTempFile(String prefix, String suffix) {
   		return Mono.fromCallable(() -> Files.createTempFile(prefix, suffix))
                .subscribeOn(Schedulers.boundedElastic());
   	}

}

There are a couple of things to note here:

1. In streaming mode, you'll have to handle the Flux<Part>, rather than individual parts.
2. The content of each part has to be consumed, for instance by calling DataBufferUtils.release() , or by calling transferTo. This makes sure the streaming parser keeps going and failure to do so will result in weird results.
3. creating a temp file is a blocking operation, and should be done on a different scheduler, like the bounded elastic scheduler.

We are considering ways to improve this rough experience in an upcoming version, allowing for code similar to your initial version, but the "read-once" attribute of streaming mode makes things quite complex.

[jomach]

I just tested this and a temporary folder is still being create :(

[keyzj]

@poutsma, hello!
Thank you for the example, but i don's see anyway that this could work. Your provided DefaultPartHttpMessageReader partReader just won't be used, because there're already registered instance of DefaultPartHttpMessageReader for MultipartHttpMessageReader.

You could enable streaming like this:

@Configuration
public class StreamingCodecConfig implements WebFluxConfigurer {
    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer config) {
        config.defaultCodecs().multipartReader(new MultipartHttpMessageReader(defaultPartHttpMessageReader()));
    }

    public DefaultPartHttpMessageReader defaultPartHttpMessageReader() {
        DefaultPartHttpMessageReader defaultPartHttpMessageReader = new DefaultPartHttpMessageReader();
        defaultPartHttpMessageReader.setStreaming(true);
        return defaultPartHttpMessageReader;
    }
}

But according to #27743 - request will hang forever (at least for me and @jomach).