org.springframework.http.ReactiveHttpOutputMessage#writeWith ( )源码实例Demo

private Mono<Void> writeMultipart(
		MultiValueMap<String, ?> map, ReactiveHttpOutputMessage outputMessage, Map<String, Object> hints) {

	byte[] boundary = generateMultipartBoundary();

	Map<String, String> params = new HashMap<>(2);
	params.put("boundary", new String(boundary, StandardCharsets.US_ASCII));
	params.put("charset", getCharset().name());

	outputMessage.getHeaders().setContentType(new MediaType(MediaType.MULTIPART_FORM_DATA, params));

	LogFormatUtils.traceDebug(logger, traceOn -> Hints.getLogPrefix(hints) + "Encoding " +
			(isEnableLoggingRequestDetails() ?
					LogFormatUtils.formatValue(map, !traceOn) :
					"parts " + map.keySet() + " (content masked)"));

	DataBufferFactory bufferFactory = outputMessage.bufferFactory();

	Flux<DataBuffer> body = Flux.fromIterable(map.entrySet())
			.concatMap(entry -> encodePartValues(boundary, entry.getKey(), entry.getValue(), bufferFactory))
			.concatWith(generateLastLine(boundary, bufferFactory))
			.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release);

	return outputMessage.writeWith(body);
}
 

@SuppressWarnings("unchecked")
@Override
public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType elementType,
		@Nullable MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {

	MediaType contentType = updateContentType(message, mediaType);

	Flux<DataBuffer> body = this.encoder.encode(
			inputStream, message.bufferFactory(), elementType, contentType, hints);

	if (inputStream instanceof Mono) {
		HttpHeaders headers = message.getHeaders();
		return Mono.from(body)
				.switchIfEmpty(Mono.defer(() -> {
					headers.setContentLength(0);
					return message.setComplete().then(Mono.empty());
				}))
				.flatMap(buffer -> {
					headers.setContentLength(buffer.readableByteCount());
					return message.writeWith(Mono.just(buffer));
				});
	}

	return (isStreamingMediaType(contentType) ?
			message.writeAndFlushWith(body.map(Flux::just)) : message.writeWith(body));
}
 

private Mono<Void> writeMultipart(
		MultiValueMap<String, ?> map, ReactiveHttpOutputMessage outputMessage, Map<String, Object> hints) {

	byte[] boundary = generateMultipartBoundary();

	Map<String, String> params = new HashMap<>(2);
	params.put("boundary", new String(boundary, StandardCharsets.US_ASCII));
	params.put("charset", getCharset().name());

	outputMessage.getHeaders().setContentType(new MediaType(MediaType.MULTIPART_FORM_DATA, params));

	LogFormatUtils.traceDebug(logger, traceOn -> Hints.getLogPrefix(hints) + "Encoding " +
			(isEnableLoggingRequestDetails() ?
					LogFormatUtils.formatValue(map, !traceOn) :
					"parts " + map.keySet() + " (content masked)"));

	Flux<DataBuffer> body = Flux.fromIterable(map.entrySet())
			.concatMap(entry -> encodePartValues(boundary, entry.getKey(), entry.getValue()))
			.concatWith(Mono.just(generateLastLine(boundary)));

	return outputMessage.writeWith(body);
}
 

@SuppressWarnings("unchecked")
@Override
public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType elementType,
		@Nullable MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {

	MediaType contentType = updateContentType(message, mediaType);

	Flux<DataBuffer> body = this.encoder.encode(
			inputStream, message.bufferFactory(), elementType, contentType, hints);

	if (inputStream instanceof Mono) {
		HttpHeaders headers = message.getHeaders();
		return Mono.from(body)
				.switchIfEmpty(Mono.defer(() -> {
					headers.setContentLength(0);
					return message.setComplete().then(Mono.empty());
				}))
				.flatMap(buffer -> {
					headers.setContentLength(buffer.readableByteCount());
					return message.writeWith(Mono.just(buffer)
							.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release));
				});
	}

	if (isStreamingMediaType(contentType)) {
		return message.writeAndFlushWith(body.map(buffer ->
				Mono.just(buffer).doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release)));
	}

	return message.writeWith(body);
}
 

private Mono<Void> encodeAndWriteRegions(Publisher<? extends ResourceRegion> publisher,
		@Nullable MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {

	Flux<DataBuffer> body = this.regionEncoder.encode(
			publisher, message.bufferFactory(), REGION_TYPE, mediaType, hints);

	return message.writeWith(body);
}
 

private Mono<Void> encodeAndWriteRegions(Publisher<? extends ResourceRegion> publisher,
		@Nullable MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {

	Flux<DataBuffer> body = this.regionEncoder.encode(
			publisher, message.bufferFactory(), REGION_TYPE, mediaType, hints);

	return message.writeWith(body);
}