private DataBuffer toDataBuffer(ContentChunk chunk) {

		// We must copy until this is resolved:
		// https://github.com/eclipse/jetty.project/issues/2429

		// Use copy instead of buffer wrapping because Callback#succeeded() is
		// used not only to release the buffer but also to request more data
		// which is a problem for codecs that buffer data.

		DataBuffer buffer = this.bufferFactory.allocateBuffer(chunk.buffer.capacity());
		buffer.write(chunk.buffer);
		chunk.callback.succeeded();
		return buffer;
	}
 

@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
	Flux<ContentChunk> chunks = Flux.from(body).map(this::toContentChunk);
	ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(chunks, getContentType());
	this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).content(content).build();
	return doCommit(this::completes);
}
 

@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
	Flux<ContentChunk> chunks = Flux.from(body)
			.flatMap(Function.identity())
			.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release)
			.map(this::toContentChunk);
	ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(chunks, getContentType());
	this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).content(content).build();
	return doCommit(this::completes);
}
 

private ContentChunk toContentChunk(DataBuffer buffer) {
	return new ContentChunk(buffer.asByteBuffer(), new Callback() {
		@Override
		public void succeeded() {
			DataBufferUtils.release(buffer);
		}

		@Override
		public void failed(Throwable x) {
			DataBufferUtils.release(buffer);
			throw Exceptions.propagate(x);
		}
	});
}
 

private DataBuffer toDataBuffer(ContentChunk chunk) {

		// We must copy until this is resolved:
		// https://github.com/eclipse/jetty.project/issues/2429

		// Use copy instead of buffer wrapping because Callback#succeeded() is
		// used not only to release the buffer but also to request more data
		// which is a problem for codecs that buffer data.

		DataBuffer buffer = this.bufferFactory.allocateBuffer(chunk.buffer.capacity());
		buffer.write(chunk.buffer);
		chunk.callback.succeeded();
		return buffer;
	}
 

@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
	Flux<ContentChunk> chunks = Flux.from(body).map(this::toContentChunk);
	ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(chunks, getContentType());
	this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).content(content).build();
	return doCommit(this::completes);
}
 

@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
	Flux<ContentChunk> chunks = Flux.from(body)
			.flatMap(Function.identity())
			.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release)
			.map(this::toContentChunk);
	ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(chunks, getContentType());
	this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).content(content).build();
	return doCommit(this::completes);
}
 

private ContentChunk toContentChunk(DataBuffer buffer) {
	return new ContentChunk(buffer.asByteBuffer(), new Callback() {
		@Override
		public void succeeded() {
			DataBufferUtils.release(buffer);
		}

		@Override
		public void failed(Throwable x) {
			DataBufferUtils.release(buffer);
			throw Exceptions.propagate(x);
		}
	});
}
 

protected ReactiveRequest.Content provideBody(ReactiveHttpRequest request) {
	Publisher<ContentChunk> bodyPublisher;
	String contentType;
	if(request.body() instanceof Mono){
		if(bodyActualClass == ByteBuffer.class){
			bodyPublisher = ((Mono)request.body()).map(this::toByteBufferChunk);
			contentType = APPLICATION_OCTET_STREAM;
		}
		else if(bodyActualClass == byte[].class){
			bodyPublisher = Flux.from(request.body()).map(this::toByteArrayChunk);
			contentType = APPLICATION_OCTET_STREAM;
		}
		else if (CharSequence.class.isAssignableFrom(bodyActualClass)){
			bodyPublisher = Flux.from(request.body()).map(this::toCharSequenceChunk);
			contentType = TEXT_UTF_8;
		}
		else {
			bodyPublisher = Flux.from(request.body()).map(data -> toJsonChunk(data, false));
			contentType = APPLICATION_JSON_UTF_8;
		}

	} else {
		if(bodyActualClass == ByteBuffer.class){
			bodyPublisher = Flux.from(request.body()).map(this::toByteBufferChunk);
			contentType = APPLICATION_OCTET_STREAM;
		}
		else if(bodyActualClass == byte[].class){
			bodyPublisher = Flux.from(request.body()).map(this::toByteArrayChunk);
			contentType = APPLICATION_OCTET_STREAM;
		}
		else {
			bodyPublisher = Flux.from(request.body()).map(data -> toJsonChunk(data, true));
			contentType = APPLICATION_STREAM_JSON_UTF_8;
		}
	}

	return ReactiveRequest.Content.fromPublisher(bodyPublisher, contentType);
}
 

protected ContentChunk toJsonChunk(Object data, boolean stream){
	try {
		ByteArrayBuilder byteArrayBuilder = new ByteArrayBuilder();
		bodyWriter.writeValue(byteArrayBuilder, data);
		if(stream) {
			byteArrayBuilder.write(NEWLINE_SEPARATOR);
		}
		ByteBuffer buffer = ByteBuffer.wrap(byteArrayBuilder.toByteArray());
		return new ContentChunk(buffer);
	} catch (java.io.IOException e) {
		throw new UncheckedIOException(e);
	}
}
 

JettyReactiveHttpResponse(Response clientResponse, Publisher<ContentChunk> contentChunks,
						  Class returnPublisherType, Class returnActualClass,
						  JsonFactory jsonFactory, ObjectReader objectReader) {
	this.clientResponse = clientResponse;
	this.contentChunks = contentChunks;
	this.returnPublisherType = returnPublisherType;
	this.returnActualClass = returnActualClass;
	this.objectReader = objectReader;
	this.jsonFactory = jsonFactory;
}
 

protected ContentChunk toByteBufferChunk(Object data){
	return new ContentChunk((ByteBuffer)data);
}
 

protected ContentChunk toByteArrayChunk(Object data){
	return new ContentChunk(ByteBuffer.wrap((byte[])data));
}
 

protected ContentChunk toCharSequenceChunk(Object data){
	CharBuffer charBuffer = CharBuffer.wrap((CharSequence) data);
	ByteBuffer byteBuffer = UTF_8.encode(charBuffer);
	return new ContentChunk(byteBuffer);
}