@SuppressWarnings("unchecked")
private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
		List<HttpMessageReader<?>> readers) {

	try {
		MediaType contentType = request.getHeaders().getContentType();
		if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
			return ((HttpMessageReader<MultiValueMap<String, Part>>) readers.stream()
					.filter(reader -> reader.canRead(MULTIPART_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
					.findFirst()
					.orElseThrow(() -> new IllegalStateException("No multipart HttpMessageReader.")))
					.readMono(MULTIPART_DATA_TYPE, request, Hints.none())
					.switchIfEmpty(EMPTY_MULTIPART_DATA)
					.cache();
		}
	}
	catch (InvalidMediaTypeException ex) {
		// Ignore
	}
	return EMPTY_MULTIPART_DATA;
}
 

public Mono<ServerResponse> multipartData(ServerRequest request) {
	return request
			.body(BodyExtractors.toMultipartData())
			.flatMap(map -> {
				Map<String, Part> parts = map.toSingleValueMap();
				try {
					assertEquals(2, parts.size());
					assertEquals("foo.txt", ((FilePart) parts.get("fooPart")).filename());
					assertEquals("bar", ((FormFieldPart) parts.get("barPart")).value());
				}
				catch(Exception e) {
					return Mono.error(e);
				}
				return ServerResponse.ok().build();
			});
}
 

@Test
public void supportsParameter() {

	MethodParameter param;

	param = this.testMethod.annot(requestPart()).arg(Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Mono.class, Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Flux.class, Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Mono.class, Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Flux.class, Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annotNotPresent(RequestPart.class).arg(Person.class);
	assertFalse(this.resolver.supportsParameter(param));
}
 

@SuppressWarnings("unchecked")
private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
		ServerCodecConfigurer configurer, String logPrefix) {

	try {
		MediaType contentType = request.getHeaders().getContentType();
		if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
			return ((HttpMessageReader<MultiValueMap<String, Part>>) configurer.getReaders().stream()
					.filter(reader -> reader.canRead(MULTIPART_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
					.findFirst()
					.orElseThrow(() -> new IllegalStateException("No multipart HttpMessageReader.")))
					.readMono(MULTIPART_DATA_TYPE, request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
					.switchIfEmpty(EMPTY_MULTIPART_DATA)
					.cache();
		}
	}
	catch (InvalidMediaTypeException ex) {
		// Ignore
	}
	return EMPTY_MULTIPART_DATA;
}
 

@RequestMapping(value = "/post", consumes = MediaType.MULTIPART_FORM_DATA_VALUE,
		produces = MediaType.APPLICATION_JSON_VALUE)
public Mono<Map<String, Object>> postFormData(
		@RequestBody Mono<MultiValueMap<String, Part>> parts) {
	// StringDecoder decoder = StringDecoder.allMimeTypes(true);
	return parts.flux().flatMap(map -> Flux.fromIterable(map.values()))
			.flatMap(Flux::fromIterable).filter(part -> part instanceof FilePart)
			.reduce(new HashMap<String, Object>(), (files, part) -> {
				MediaType contentType = part.headers().getContentType();
				long contentLength = part.headers().getContentLength();
				// TODO: get part data
				files.put(part.name(),
						"data:" + contentType + ";base64," + contentLength);
				return files;
			}).map(files -> Collections.singletonMap("files", files));
}
 

/**
 * Add an asynchronous part with {@link Publisher}-based content.
 * @param name the name of the part to add
 * @param publisher a Publisher of content for the part
 * @param elementClass the type of elements contained in the publisher
 * @return builder that allows for further customization of part headers
 */
@SuppressWarnings("unchecked")
public <T, P extends Publisher<T>> PartBuilder asyncPart(String name, P publisher, Class<T> elementClass) {
	Assert.hasLength(name, "'name' must not be empty");
	Assert.notNull(publisher, "'publisher' must not be null");
	Assert.notNull(elementClass, "'elementClass' must not be null");

	if (Part.class.isAssignableFrom(elementClass)) {
		publisher = (P) Mono.from(publisher).flatMapMany(p -> ((Part) p).content());
		elementClass = (Class<T>) DataBuffer.class;
	}

	HttpHeaders headers = new HttpHeaders();
	PublisherPartBuilder<T, P> builder = new PublisherPartBuilder<>(headers, publisher, elementClass);
	this.parts.add(name, builder);
	return builder;

}
 

@SaveAction
@PostMapping("/upload")
@SneakyThrows
public Mono<String> upload(@RequestPart("file") Part part) {

    if (part instanceof FilePart) {
        return (part)
                .content()
                .collectList()
                .flatMap(all -> Mono.fromCallable(() ->
                        Base64.encodeBase64String(StreamUtils.copyToByteArray(factory.join(all).asInputStream()))))
                ;
    } else {
        return Mono.error(() -> new IllegalArgumentException("[file] part is not a file"));
    }

}
 

@SuppressWarnings("unchecked")
private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
		List<HttpMessageReader<?>> readers) {

	try {
		MediaType contentType = request.getHeaders().getContentType();
		if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
			return ((HttpMessageReader<MultiValueMap<String, Part>>) readers.stream()
					.filter(reader -> reader.canRead(MULTIPART_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
					.findFirst()
					.orElseThrow(() -> new IllegalStateException("No multipart HttpMessageReader.")))
					.readMono(MULTIPART_DATA_TYPE, request, Hints.none())
					.switchIfEmpty(EMPTY_MULTIPART_DATA)
					.cache();
		}
	}
	catch (InvalidMediaTypeException ex) {
		// Ignore
	}
	return EMPTY_MULTIPART_DATA;
}
 

public Mono<ServerResponse> multipartData(ServerRequest request) {
	return request
			.body(BodyExtractors.toMultipartData())
			.flatMap(map -> {
				Map<String, Part> parts = map.toSingleValueMap();
				try {
					assertEquals(2, parts.size());
					assertEquals("foo.txt", ((FilePart) parts.get("fooPart")).filename());
					assertEquals("bar", ((FormFieldPart) parts.get("barPart")).value());
				}
				catch(Exception e) {
					return Mono.error(e);
				}
				return ServerResponse.ok().build();
			});
}
 

@Test
public void supportsParameter() {

	MethodParameter param;

	param = this.testMethod.annot(requestPart()).arg(Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Mono.class, Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Flux.class, Person.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Mono.class, Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annot(requestPart()).arg(Flux.class, Part.class);
	assertTrue(this.resolver.supportsParameter(param));

	param = this.testMethod.annotNotPresent(RequestPart.class).arg(Person.class);
	assertFalse(this.resolver.supportsParameter(param));
}
 

@SuppressWarnings("unchecked")
private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
		ServerCodecConfigurer configurer, String logPrefix) {

	try {
		MediaType contentType = request.getHeaders().getContentType();
		if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
			return ((HttpMessageReader<MultiValueMap<String, Part>>) configurer.getReaders().stream()
					.filter(reader -> reader.canRead(MULTIPART_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
					.findFirst()
					.orElseThrow(() -> new IllegalStateException("No multipart HttpMessageReader.")))
					.readMono(MULTIPART_DATA_TYPE, request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
					.switchIfEmpty(EMPTY_MULTIPART_DATA)
					.cache();
		}
	}
	catch (InvalidMediaTypeException ex) {
		// Ignore
	}
	return EMPTY_MULTIPART_DATA;
}
 

public static Mono<Map<String, Object>> extractValuesToBind(ServerRequest request) {
    MultiValueMap<String, String> queryParams = request.queryParams();
    Mono<MultiValueMap<String, String>> formData = request.formData();
    Mono<MultiValueMap<String, Part>> multipartData = request.multipartData();

    return Mono.zip(Mono.just(queryParams), formData, multipartData)
            .map(tuple -> {
                Map<String, Object> result = new TreeMap<>();
                tuple.getT1().forEach((key, values) -> addBindValue(result, key, values));
                tuple.getT2().forEach((key, values) -> addBindValue(result, key, values));
                tuple.getT3().forEach((key, values) -> addBindValue(result, key, values));
                return result;
            });
}
 

/**
 * Extractor to read multipart data into a {@code MultiValueMap<String, Part>}.
 * @return {@code BodyExtractor} for multipart data
 */
// Parameterized for server-side use
public static BodyExtractor<Mono<MultiValueMap<String, Part>>, ServerHttpRequest> toMultipartData() {
	return (serverRequest, context) -> {
		ResolvableType elementType = MULTIPART_DATA_TYPE;
		MediaType mediaType = MediaType.MULTIPART_FORM_DATA;
		HttpMessageReader<MultiValueMap<String, Part>> reader = findReader(elementType, mediaType, context);
		return readToMono(serverRequest, context, elementType, reader);
	};
}
 

/**
 * Extractor to read multipart data into {@code Flux<Part>}.
 * @return {@code BodyExtractor} for multipart request parts
 */
// Parameterized for server-side use
public static BodyExtractor<Flux<Part>, ServerHttpRequest> toParts() {
	return (serverRequest, context) -> {
		ResolvableType elementType = PART_TYPE;
		MediaType mediaType = MediaType.MULTIPART_FORM_DATA;
		HttpMessageReader<Part> reader = findReader(elementType, mediaType, context);
		return readToFlux(serverRequest, context, elementType, reader);
	};
}
 

/**
 * Multipart file upload
 * @param bucket
 * @param parts
 * @param headers
 * @return
 */
@RequestMapping(consumes = MediaType.MULTIPART_FORM_DATA_VALUE, method = {RequestMethod.POST, RequestMethod.PUT})
public Mono<ResponseEntity<UploadResult>> multipartUploadHandler(@RequestHeader HttpHeaders headers, @RequestBody Flux<Part> parts  ) {
            
    return parts
      .ofType(FilePart.class) // We'll ignore other data for now
      .flatMap((part) -> saveFile(headers, s3config.getBucket(), part))
      .collect(Collectors.toList())
      .map((keys) -> ResponseEntity.status(HttpStatus.CREATED)
        .body(new UploadResult(HttpStatus.CREATED,keys)));
}
 

@Test
public void part() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	Part actual = resolveArgument(param, bodyBuilder);

	DataBuffer buffer = DataBufferUtils.join(actual.content()).block();
	assertEquals("{\"name\":\"Jones\"}", DataBufferTestUtils.dumpString(buffer, StandardCharsets.UTF_8));
}
 

@Test
public void listPart() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(List.class, Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	bodyBuilder.part("name", new Person("James"));
	List<Part> actual = resolveArgument(param, bodyBuilder);

	assertEquals("{\"name\":\"Jones\"}", partToUtf8String(actual.get(0)));
	assertEquals("{\"name\":\"James\"}", partToUtf8String(actual.get(1)));
}
 

@Test
public void fluxPart() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(Flux.class, Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	bodyBuilder.part("name", new Person("James"));
	Flux<Part> actual = resolveArgument(param, bodyBuilder);

	List<Part> parts = actual.collectList().block();
	assertEquals("{\"name\":\"Jones\"}", partToUtf8String(parts.get(0)));
	assertEquals("{\"name\":\"James\"}", partToUtf8String(parts.get(1)));
}
 

@Test
public void partRequired() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(Part.class);
	ServerWebExchange exchange = createExchange(new MultipartBodyBuilder());
	Mono<Object> result = this.resolver.resolveArgument(param, new BindingContext(), exchange);

	StepVerifier.create(result).expectError(ServerWebInputException.class).verify();
}
 

@Test
public void partNotRequired() {
	MethodParameter param = this.testMethod.annot(requestPart().notRequired()).arg(Part.class);
	ServerWebExchange exchange = createExchange(new MultipartBodyBuilder());
	Mono<Object> result = this.resolver.resolveArgument(param, new BindingContext(), exchange);

	StepVerifier.create(result).verifyComplete();
}
 

@SuppressWarnings("unused")
void handle(
		@RequestPart("name") Person person,
		@RequestPart("name") Mono<Person> personMono,
		@RequestPart("name") Flux<Person> personFlux,
		@RequestPart("name") List<Person> personList,
		@RequestPart("name") Part part,
		@RequestPart("name") Mono<Part> partMono,
		@RequestPart("name") Flux<Part> partFlux,
		@RequestPart("name") List<Part> partList,
		@RequestPart(name = "anotherPart", required = false) Person anotherPerson,
		@RequestPart(name = "anotherPart", required = false) Part anotherPart,
		Person notAnnotated) {}
 

/**
 * Combine query params and form data for multipart form data from the body
 * of the request into a {@code Map<String, Object>} of values to use for
 * data binding purposes.
 * @param exchange the current exchange
 * @return a {@code Mono} with the values to bind
 * @see org.springframework.http.server.reactive.ServerHttpRequest#getQueryParams()
 * @see ServerWebExchange#getFormData()
 * @see ServerWebExchange#getMultipartData()
 */
public static Mono<Map<String, Object>> extractValuesToBind(ServerWebExchange exchange) {
	MultiValueMap<String, String> queryParams = exchange.getRequest().getQueryParams();
	Mono<MultiValueMap<String, String>> formData = exchange.getFormData();
	Mono<MultiValueMap<String, Part>> multipartData = exchange.getMultipartData();

	return Mono.zip(Mono.just(queryParams), formData, multipartData)
			.map(tuple -> {
				Map<String, Object> result = new TreeMap<>();
				tuple.getT1().forEach((key, values) -> addBindValue(result, key, values));
				tuple.getT2().forEach((key, values) -> addBindValue(result, key, values));
				tuple.getT3().forEach((key, values) -> addBindValue(result, key, values));
				return result;
			});
}
 

@Override
protected void extendTypedReaders(List<HttpMessageReader<?>> typedReaders) {

	HttpMessageReader<Part> partReader = getMultipartReader();

	boolean logRequestDetails = isEnableLoggingRequestDetails();
	if (partReader instanceof LoggingCodecSupport) {
		((LoggingCodecSupport) partReader).setEnableLoggingRequestDetails(logRequestDetails);
	}
	typedReaders.add(partReader);

	MultipartHttpMessageReader reader = new MultipartHttpMessageReader(partReader);
	reader.setEnableLoggingRequestDetails(logRequestDetails);
	typedReaders.add(reader);
}
 

private void assertFooPart(Part part) {
	assertEquals("fooPart", part.name());
	assertTrue(part instanceof FilePart);
	assertEquals("foo.txt", ((FilePart) part).filename());

	StepVerifier.create(DataBufferUtils.join(part.content()))
			.consumeNextWith(buffer -> {
				assertEquals(12, buffer.readableByteCount());
				byte[] byteContent = new byte[12];
				buffer.read(byteContent);
				assertEquals("Lorem Ipsum.", new String(byteContent));
			})
			.verifyComplete();
}
 

/**
 * Extractor to read multipart data into a {@code MultiValueMap<String, Part>}.
 * @return {@code BodyExtractor} for multipart data
 */
// Parameterized for server-side use
public static BodyExtractor<Mono<MultiValueMap<String, Part>>, ServerHttpRequest> toMultipartData() {
	return (serverRequest, context) -> {
		ResolvableType elementType = MULTIPART_DATA_TYPE;
		MediaType mediaType = MediaType.MULTIPART_FORM_DATA;
		HttpMessageReader<MultiValueMap<String, Part>> reader = findReader(elementType, mediaType, context);
		return readToMono(serverRequest, context, elementType, reader);
	};
}
 

/**
 * Extractor to read multipart data into {@code Flux<Part>}.
 * @return {@code BodyExtractor} for multipart request parts
 */
// Parameterized for server-side use
public static BodyExtractor<Flux<Part>, ServerHttpRequest> toParts() {
	return (serverRequest, context) -> {
		ResolvableType elementType = PART_TYPE;
		MediaType mediaType = MediaType.MULTIPART_FORM_DATA;
		HttpMessageReader<Part> reader = findReader(elementType, mediaType, context);
		return readToFlux(serverRequest, context, elementType, reader);
	};
}
 

private Flux<?> decodePartValues(Flux<Part> parts, MethodParameter elementType, BindingContext bindingContext,
		ServerWebExchange exchange, boolean isRequired) {

	return parts.flatMap(part -> {
		ServerHttpRequest partRequest = new PartServerHttpRequest(exchange.getRequest(), part);
		ServerWebExchange partExchange = exchange.mutate().request(partRequest).build();
		if (logger.isDebugEnabled()) {
			logger.debug(exchange.getLogPrefix() + "Decoding part '" + part.name() + "'");
		}
		return readBody(elementType, isRequired, bindingContext, partExchange);
	});
}
 

@Test
public void part() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	Part actual = resolveArgument(param, bodyBuilder);

	DataBuffer buffer = DataBufferUtils.join(actual.content()).block();
	assertEquals("{\"name\":\"Jones\"}", DataBufferTestUtils.dumpString(buffer, StandardCharsets.UTF_8));
}
 

@Test
public void listPart() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(List.class, Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	bodyBuilder.part("name", new Person("James"));
	List<Part> actual = resolveArgument(param, bodyBuilder);

	assertEquals("{\"name\":\"Jones\"}", partToUtf8String(actual.get(0)));
	assertEquals("{\"name\":\"James\"}", partToUtf8String(actual.get(1)));
}
 

@Test
public void fluxPart() {
	MethodParameter param = this.testMethod.annot(requestPart()).arg(Flux.class, Part.class);
	MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
	bodyBuilder.part("name", new Person("Jones"));
	bodyBuilder.part("name", new Person("James"));
	Flux<Part> actual = resolveArgument(param, bodyBuilder);

	List<Part> parts = actual.collectList().block();
	assertEquals("{\"name\":\"Jones\"}", partToUtf8String(parts.get(0)));
	assertEquals("{\"name\":\"James\"}", partToUtf8String(parts.get(1)));
}