Merge pull request #63 from poutsma/byte_buffer_abstraction

Byte buffer abstraction
This commit is contained in:
Arjen Poutsma
2016-01-27 16:17:04 +01:00
67 changed files with 1909 additions and 646 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,13 +16,13 @@
package org.springframework.core.codec;
import java.nio.ByteBuffer;
import java.util.List;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.MimeType;
/**
@@ -43,14 +43,14 @@ public interface Decoder<T> {
boolean canDecode(ResolvableType type, MimeType mimeType, Object... hints);
/**
* Decode an input {@link ByteBuffer} stream to an output stream of {@code T}.
* Decode an input {@link DataBuffer} stream to an output stream of {@code T}.
* @param inputStream the input stream to process.
* @param type the stream element type to process.
* @param mimeType the mime type to process.
* @param hints Additional information about how to do decode, optional.
* @return the output stream
*/
Flux<T> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
Flux<T> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints);
/**

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.MimeType;
/**
@@ -50,7 +51,7 @@ public interface Encoder<T> {
* @param hints Additional information about how to do decode, optional.
* @return the output stream
*/
Flux<ByteBuffer> encode(Publisher<? extends T> inputStream, ResolvableType type,
Flux<DataBuffer> encode(Publisher<? extends T> inputStream, ResolvableType type,
MimeType mimeType, Object... hints);
/**

View File

@@ -0,0 +1,42 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.codec.support;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
* @author Arjen Poutsma
*/
public abstract class AbstractAllocatingEncoder<T> extends AbstractEncoder<T> {
private final DataBufferAllocator allocator;
public AbstractAllocatingEncoder(DataBufferAllocator allocator,
MimeType... supportedMimeTypes) {
super(supportedMimeTypes);
Assert.notNull(allocator, "'allocator' must not be null");
this.allocator = allocator;
}
public DataBufferAllocator allocator() {
return allocator;
}
}

View File

@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
@@ -28,6 +27,9 @@ import reactor.core.util.BackpressureUtils;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
@@ -38,12 +40,19 @@ import org.springframework.util.MimeType;
*/
public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T> {
public AbstractRawByteStreamDecoder(MimeType... supportedMimeTypes) {
private final DataBufferAllocator allocator;
public AbstractRawByteStreamDecoder(DataBufferAllocator allocator,
MimeType... supportedMimeTypes) {
super(supportedMimeTypes);
Assert.notNull(allocator, "'allocator' must not be null");
this.allocator = allocator;
}
@Override
public Flux<T> decode(Publisher<ByteBuffer> inputStream, ResolvableType type, MimeType mimeType, Object... hints) {
public Flux<T> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints) {
return decodeInternal(Flux.from(inputStream).lift(bbs -> subscriberBarrier(bbs)),
type, mimeType, hints);
@@ -55,17 +64,20 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
* <p>Implementations should provide their own {@link SubscriberBarrier} or use one of the
* provided implementations by this class
*/
public abstract SubscriberBarrier<ByteBuffer, ByteBuffer> subscriberBarrier(Subscriber<? super ByteBuffer> subscriber);
public abstract SubscriberBarrier<DataBuffer, DataBuffer> subscriberBarrier(
Subscriber<? super DataBuffer> subscriber);
public abstract Flux<T> decodeInternal(Publisher<ByteBuffer> inputStream, ResolvableType type
public abstract Flux<T> decodeInternal(Publisher<DataBuffer> inputStream,
ResolvableType type
, MimeType mimeType, Object... hints);
/**
* {@code SubscriberBarrier} implementation that buffers all received elements and emits a single
* {@code ByteBuffer} once the incoming stream has been completed
* {@code DataBuffer} once the incoming stream has been completed
*/
public static class ReduceSingleByteStreamBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
public static class ReduceSingleByteStreamBarrier
extends SubscriberBarrier<DataBuffer, DataBuffer> {
@SuppressWarnings("rawtypes")
static final AtomicLongFieldUpdater<ReduceSingleByteStreamBarrier> REQUESTED =
@@ -74,16 +86,16 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
static final AtomicIntegerFieldUpdater<ReduceSingleByteStreamBarrier> TERMINATED =
AtomicIntegerFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "terminated");
private volatile long requested;
private volatile int terminated;
private ByteBuffer buffer;
private DataBuffer buffer;
public ReduceSingleByteStreamBarrier(Subscriber<? super ByteBuffer> subscriber) {
public ReduceSingleByteStreamBarrier(Subscriber<? super DataBuffer> subscriber,
DataBufferAllocator allocator) {
super(subscriber);
this.buffer = ByteBuffer.allocate(0);
this.buffer = allocator.allocateBuffer();
}
@Override
@@ -108,15 +120,12 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
* TODO: when available, wrap buffers with a single buffer and avoid copying data for every method call.
*/
@Override
protected void doNext(ByteBuffer byteBuffer) {
this.buffer = ByteBuffer.allocate(this.buffer.capacity() + byteBuffer.capacity())
.put(this.buffer).put(byteBuffer);
this.buffer.flip();
protected void doNext(DataBuffer dataBuffer) {
this.buffer.write(dataBuffer);
}
protected void drainLast() {
if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) {
this.buffer.flip();
subscriber.onNext(this.buffer);
super.doComplete();
}
@@ -127,7 +136,8 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
* {@code SubscriberBarrier} implementation that splits incoming elements
* using line return delimiters: {@code "\n"} and {@code "\r\n"}
*/
public static class SplitLinesByteStreamBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
public static class SplitLinesByteStreamBarrier
extends SubscriberBarrier<DataBuffer, DataBuffer> {
@SuppressWarnings("rawtypes")
static final AtomicLongFieldUpdater<SplitLinesByteStreamBarrier> REQUESTED =
@@ -136,16 +146,20 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
static final AtomicIntegerFieldUpdater<SplitLinesByteStreamBarrier> TERMINATED =
AtomicIntegerFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "terminated");
private final DataBufferAllocator allocator;
private volatile long requested;
private volatile int terminated;
private ByteBuffer buffer;
private DataBuffer buffer;
public SplitLinesByteStreamBarrier(Subscriber<? super ByteBuffer> subscriber) {
public SplitLinesByteStreamBarrier(Subscriber<? super DataBuffer> subscriber,
DataBufferAllocator allocator) {
super(subscriber);
this.buffer = ByteBuffer.allocate(0);
this.allocator = allocator;
this.buffer = allocator.allocateBuffer();
}
@Override
@@ -170,19 +184,20 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
* TODO: when available, wrap buffers with a single buffer and avoid copying data for every method call.
*/
@Override
protected void doNext(ByteBuffer byteBuffer) {
this.buffer = ByteBuffer.allocate(this.buffer.capacity() + byteBuffer.capacity())
.put(this.buffer).put(byteBuffer);
protected void doNext(DataBuffer dataBuffer) {
this.buffer.write(dataBuffer);
while (REQUESTED.get(this) > 0) {
int separatorIndex = findEndOfLine(this.buffer);
if (separatorIndex != -1) {
if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) {
byte[] message = new byte[separatorIndex];
this.buffer.get(message);
this.buffer.read(message);
consumeSeparator(this.buffer);
this.buffer = this.buffer.slice();
super.doNext(ByteBuffer.wrap(message));
// this.buffer = this.buffer.slice();
DataBuffer buffer2 = allocator.allocateBuffer(message.length);
buffer2.write(message);
super.doNext(buffer2);
}
}
else {
@@ -191,9 +206,9 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
}
}
protected int findEndOfLine(ByteBuffer buffer) {
protected int findEndOfLine(DataBuffer buffer) {
final int n = buffer.limit();
final int n = buffer.readableByteCount();
for (int i = 0; i < n; i++) {
final byte b = buffer.get(i);
if (b == '\n') {
@@ -207,16 +222,15 @@ public abstract class AbstractRawByteStreamDecoder<T> extends AbstractDecoder<T>
return -1;
}
protected void consumeSeparator(ByteBuffer buffer) {
byte sep = buffer.get();
protected void consumeSeparator(DataBuffer buffer) {
byte sep = buffer.read();
if (sep == '\r') {
buffer.get();
buffer.read();
}
}
protected void drainLast() {
if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) {
this.buffer.flip();
subscriber.onNext(this.buffer);
super.doComplete();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -43,10 +44,9 @@ public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
}
@Override
public Flux<ByteBuffer> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
public Flux<ByteBuffer> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints) {
return Flux.from(inputStream);
return Flux.from(inputStream).map(DataBuffer::asByteBuffer);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,17 +22,18 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
/**
* @author Sebastien Deleuze
*/
public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
public class ByteBufferEncoder extends AbstractAllocatingEncoder<ByteBuffer> {
public ByteBufferEncoder() {
super(MimeTypeUtils.ALL);
public ByteBufferEncoder(DataBufferAllocator allocator) {
super(allocator, MimeTypeUtils.ALL);
}
@@ -43,11 +44,16 @@ public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
}
@Override
public Flux<ByteBuffer> encode(Publisher<? extends ByteBuffer> inputStream, ResolvableType type,
public Flux<DataBuffer> encode(Publisher<? extends ByteBuffer> inputStream,
ResolvableType type,
MimeType mimeType, Object... hints) {
//noinspection unchecked
return Flux.from(inputStream);
return Flux.from(inputStream).map(byteBuffer -> {
DataBuffer dataBuffer = allocator().allocateBuffer(byteBuffer.remaining());
dataBuffer.write(byteBuffer);
return dataBuffer;
});
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,7 +17,6 @@
package org.springframework.core.codec.support;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -28,7 +27,7 @@ import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.Decoder;
import org.springframework.util.ByteBufferInputStream;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.MimeType;
@@ -42,39 +41,38 @@ public class JacksonJsonDecoder extends AbstractDecoder<Object> {
private final ObjectMapper mapper;
private Decoder<ByteBuffer> preProcessor;
private Decoder<DataBuffer> preProcessor;
public JacksonJsonDecoder() {
this(new ObjectMapper(), null);
}
public JacksonJsonDecoder(Decoder<ByteBuffer> preProcessor) {
public JacksonJsonDecoder(Decoder<DataBuffer> preProcessor) {
this(new ObjectMapper(), preProcessor);
}
public JacksonJsonDecoder(ObjectMapper mapper, Decoder<ByteBuffer> preProcessor) {
public JacksonJsonDecoder(ObjectMapper mapper, Decoder<DataBuffer> preProcessor) {
super(new MimeType("application", "json", StandardCharsets.UTF_8),
new MimeType("application", "*+json", StandardCharsets.UTF_8));
this.mapper = mapper;
this.preProcessor = preProcessor;
}
@Override
public Flux<Object> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints) {
ObjectReader reader = this.mapper.readerFor(type.getRawClass());
Flux<ByteBuffer> stream = Flux.from(inputStream);
Flux<DataBuffer> stream = Flux.from(inputStream);
if (this.preProcessor != null) {
stream = this.preProcessor.decode(inputStream, type, mimeType, hints);
}
return stream.map(content -> {
try {
return reader.readValue(new ByteBufferInputStream(content));
return reader.readValue(content.asInputStream());
}
catch (IOException e) {
throw new CodecException("Error while reading the data", e);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,19 +17,19 @@
package org.springframework.core.codec.support;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.Encoder;
import org.springframework.util.BufferOutputStream;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
/**
@@ -38,50 +38,49 @@ import org.springframework.util.MimeType;
* @author Sebastien Deleuze
* @see JacksonJsonDecoder
*/
public class JacksonJsonEncoder extends AbstractEncoder<Object> {
public class JacksonJsonEncoder extends AbstractAllocatingEncoder<Object> {
private final ObjectMapper mapper;
private Encoder<ByteBuffer> postProcessor;
private Encoder<DataBuffer> postProcessor;
public JacksonJsonEncoder() {
this(new ObjectMapper(), null);
public JacksonJsonEncoder(DataBufferAllocator allocator) {
this(allocator, new ObjectMapper(), null);
}
public JacksonJsonEncoder(Encoder<ByteBuffer> postProcessor) {
this(new ObjectMapper(), postProcessor);
public JacksonJsonEncoder(DataBufferAllocator allocator,
Encoder<DataBuffer> postProcessor) {
this(allocator, new ObjectMapper(), postProcessor);
}
public JacksonJsonEncoder(ObjectMapper mapper, Encoder<ByteBuffer> postProcessor) {
super(new MimeType("application", "json", StandardCharsets.UTF_8),
public JacksonJsonEncoder(DataBufferAllocator allocator, ObjectMapper mapper,
Encoder<DataBuffer> postProcessor) {
super(allocator, new MimeType("application", "json", StandardCharsets.UTF_8),
new MimeType("application", "*+json", StandardCharsets.UTF_8));
this.mapper = mapper;
this.postProcessor = postProcessor;
}
@Override
public Flux<ByteBuffer> encode(Publisher<? extends Object> inputStream,
public Flux<DataBuffer> encode(Publisher<? extends Object> inputStream,
ResolvableType type, MimeType mimeType, Object... hints) {
Publisher<ByteBuffer> stream = (inputStream instanceof Mono ?
Publisher<DataBuffer> stream = (inputStream instanceof Mono ?
((Mono<?>)inputStream).map(this::serialize) :
Flux.from(inputStream).map(this::serialize));
return (this.postProcessor == null ? Flux.from(stream) : this.postProcessor.encode(stream, type, mimeType, hints));
}
private ByteBuffer serialize(Object value) {
Buffer buffer = new Buffer();
BufferOutputStream outputStream = new BufferOutputStream(buffer);
private DataBuffer serialize(Object value) {
DataBuffer buffer = allocator().allocateBuffer();
OutputStream outputStream = buffer.asOutputStream();
try {
this.mapper.writeValue(outputStream, value);
}
catch (IOException e) {
throw new CodecException("Error while writing the data", e);
}
buffer.flip();
return buffer.byteBuffer();
return buffer;
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import javax.xml.bind.JAXBContext;
@@ -38,8 +37,9 @@ import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.util.Assert;
import org.springframework.util.ByteBufferPublisherInputStream;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -60,12 +60,13 @@ public class Jaxb2Decoder extends AbstractDecoder<Object> {
@Override
public Flux<Object> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints) {
Class<?> outputClass = type.getRawClass();
try {
Source source = processSource(new StreamSource(new ByteBufferPublisherInputStream(inputStream)));
Source source = processSource(
new StreamSource(DataBufferUtils.toInputStream(inputStream)));
Unmarshaller unmarshaller = createUnmarshaller(outputClass);
if (outputClass.isAnnotationPresent(XmlRootElement.class)) {
return Flux.just(unmarshaller.unmarshal(source));

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -27,12 +27,12 @@ import javax.xml.bind.Marshaller;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.Assert;
import org.springframework.util.BufferOutputStream;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -43,30 +43,29 @@ import org.springframework.util.MimeTypeUtils;
* @author Sebastien Deleuze
* @see Jaxb2Decoder
*/
public class Jaxb2Encoder extends AbstractEncoder<Object> {
public class Jaxb2Encoder extends AbstractAllocatingEncoder<Object> {
private final ConcurrentMap<Class<?>, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64);
public Jaxb2Encoder() {
super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
public Jaxb2Encoder(DataBufferAllocator allocator) {
super(allocator, MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
}
@Override
public Flux<ByteBuffer> encode(Publisher<? extends Object> messageStream, ResolvableType type,
public Flux<DataBuffer> encode(Publisher<? extends Object> messageStream,
ResolvableType type,
MimeType mimeType, Object... hints) {
return Flux.from(messageStream).map(value -> {
try {
Buffer buffer = new Buffer();
BufferOutputStream outputStream = new BufferOutputStream(buffer);
DataBuffer buffer = allocator().allocateBuffer(1024);
OutputStream outputStream = buffer.asOutputStream();
Class<?> clazz = ClassUtils.getUserClass(value);
Marshaller marshaller = createMarshaller(clazz);
marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name());
marshaller.marshal(value, outputStream);
buffer.flip();
return buffer.byteBuffer();
return buffer;
}
catch (MarshalException ex) {
throw new CodecException("Could not marshal [" + value + "]: " + ex.getMessage(), ex);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
@@ -29,6 +28,8 @@ import reactor.core.publisher.Flux;
import reactor.fn.Function;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
/**
@@ -44,7 +45,7 @@ import org.springframework.util.MimeType;
* @author Sebastien Deleuze
* @see JsonObjectEncoder
*/
public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
public class JsonObjectDecoder extends AbstractDecoder<DataBuffer> {
private static final int ST_CORRUPTED = -1;
@@ -54,38 +55,40 @@ public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
private static final int ST_DECODING_ARRAY_STREAM = 2;
private final DataBufferAllocator allocator;
private final int maxObjectLength;
private final boolean streamArrayElements;
public JsonObjectDecoder() {
public JsonObjectDecoder(DataBufferAllocator allocator) {
// 1 MB
this(1024 * 1024);
this(allocator, 1024 * 1024);
}
public JsonObjectDecoder(int maxObjectLength) {
this(maxObjectLength, true);
public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength) {
this(allocator, maxObjectLength, true);
}
public JsonObjectDecoder(boolean streamArrayElements) {
this(1024 * 1024, streamArrayElements);
public JsonObjectDecoder(DataBufferAllocator allocator, boolean streamArrayElements) {
this(allocator, 1024 * 1024, streamArrayElements);
}
/**
* @param allocator
* @param maxObjectLength maximum number of bytes a JSON object/array may
* use (including braces and all). Objects exceeding this length are dropped
* and an {@link IllegalStateException} is thrown.
* @param streamArrayElements if set to true and the "top level" JSON object
* is an array, each of its entries is passed through the pipeline individually
* and immediately after it was fully received, allowing for arrays with
* "infinitely" many elements.
*/
public JsonObjectDecoder(int maxObjectLength, boolean streamArrayElements) {
public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength,
boolean streamArrayElements) {
super(new MimeType("application", "json", StandardCharsets.UTF_8),
new MimeType("application", "*+json", StandardCharsets.UTF_8));
this.allocator = allocator;
if (maxObjectLength < 1) {
throw new IllegalArgumentException("maxObjectLength must be a positive int");
}
@@ -94,10 +97,11 @@ public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
}
@Override
public Flux<ByteBuffer> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
public Flux<DataBuffer> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
MimeType mimeType, Object... hints) {
return Flux.from(inputStream).flatMap(new Function<ByteBuffer, Publisher<? extends ByteBuffer>>() {
return Flux.from(inputStream)
.flatMap(new Function<DataBuffer, Publisher<? extends DataBuffer>>() {
int openBraces;
int index;
@@ -107,14 +111,15 @@ public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
Integer writerIndex;
@Override
public Publisher<? extends ByteBuffer> apply(ByteBuffer b) {
List<ByteBuffer> chunks = new ArrayList<>();
public Publisher<? extends DataBuffer> apply(DataBuffer b) {
List<DataBuffer> chunks = new ArrayList<>();
if (this.input == null) {
this.input = Unpooled.copiedBuffer(b);
this.input = Unpooled.copiedBuffer(b.asByteBuffer());
this.writerIndex = this.input.writerIndex();
}
else {
this.input = Unpooled.copiedBuffer(this.input, Unpooled.copiedBuffer(b));
this.input = Unpooled.copiedBuffer(this.input,
Unpooled.copiedBuffer(b.asByteBuffer()));
this.writerIndex = this.input.writerIndex();
}
if (this.state == ST_CORRUPTED) {
@@ -139,7 +144,7 @@ public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
ByteBuf json = extractObject(this.input, this.input.readerIndex(),
this.index + 1 - this.input.readerIndex());
if (json != null) {
chunks.add(json.nioBuffer());
chunks.add(allocator.wrap(json.nioBuffer()));
}
// The JSON object/array was extracted => discard the bytes from
@@ -173,7 +178,7 @@ public class JsonObjectDecoder extends AbstractDecoder<ByteBuffer> {
idxNoSpaces + 1 - this.input.readerIndex());
if (json != null) {
chunks.add(json.nioBuffer());
chunks.add(allocator.wrap(json.nioBuffer()));
}
this.input.readerIndex(this.index + 1);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
@@ -27,9 +26,10 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.subscriber.SubscriberBarrier;
import reactor.core.util.BackpressureUtils;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
/**
@@ -42,25 +42,25 @@ import org.springframework.util.MimeType;
*
* @see JsonObjectDecoder
*/
public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
public class JsonObjectEncoder extends AbstractAllocatingEncoder<DataBuffer> {
public JsonObjectEncoder() {
super(new MimeType("application", "json", StandardCharsets.UTF_8),
public JsonObjectEncoder(DataBufferAllocator allocator) {
super(allocator, new MimeType("application", "json", StandardCharsets.UTF_8),
new MimeType("application", "*+json", StandardCharsets.UTF_8));
}
@Override
public Flux<ByteBuffer> encode(Publisher<? extends ByteBuffer> inputStream,
public Flux<DataBuffer> encode(Publisher<? extends DataBuffer> inputStream,
ResolvableType type, MimeType mimeType, Object... hints) {
if (inputStream instanceof Mono) {
return Flux.from(inputStream);
}
return Flux.from(inputStream).lift(s -> new JsonArrayEncoderBarrier(s));
return Flux.from(inputStream)
.lift(s -> new JsonArrayEncoderBarrier(s, allocator()));
}
private static class JsonArrayEncoderBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
private static class JsonArrayEncoderBarrier
extends SubscriberBarrier<DataBuffer, DataBuffer> {
@SuppressWarnings("rawtypes")
static final AtomicLongFieldUpdater<JsonArrayEncoderBarrier> REQUESTED =
@@ -69,8 +69,9 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
static final AtomicIntegerFieldUpdater<JsonArrayEncoderBarrier> TERMINATED =
AtomicIntegerFieldUpdater.newUpdater(JsonArrayEncoderBarrier.class, "terminated");
private final DataBufferAllocator allocator;
private ByteBuffer prev = null;
private DataBuffer prev = null;
private long count = 0;
@@ -78,9 +79,10 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
private volatile int terminated;
public JsonArrayEncoderBarrier(Subscriber<? super ByteBuffer> subscriber) {
public JsonArrayEncoderBarrier(Subscriber<? super DataBuffer> subscriber,
DataBufferAllocator allocator) {
super(subscriber);
this.allocator = allocator;
}
@@ -96,34 +98,32 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
}
@Override
protected void doNext(ByteBuffer next) {
protected void doNext(DataBuffer next) {
this.count++;
ByteBuffer tmp = this.prev;
DataBuffer tmp = this.prev;
this.prev = next;
Buffer buffer = new Buffer();
DataBuffer buffer = allocator.allocateBuffer();
if (this.count == 1) {
buffer.append("[");
buffer.write((byte) '[');
}
if (tmp != null) {
buffer.append(tmp);
buffer.write(tmp);
}
if (this.count > 1) {
buffer.append(",");
buffer.write((byte) ',');
}
buffer.flip();
BackpressureUtils.getAndSub(REQUESTED, this, 1L);
subscriber.onNext(buffer.byteBuffer());
subscriber.onNext(buffer);
}
protected void drainLast(){
if(BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) {
Buffer buffer = new Buffer();
buffer.append(this.prev);
buffer.append("]");
buffer.flip();
subscriber.onNext(buffer.byteBuffer());
DataBuffer buffer = allocator.allocateBuffer();
buffer.write(this.prev);
buffer.write((byte) ']');
subscriber.onNext(buffer);
super.doComplete();
}
}

View File

@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@@ -26,6 +25,8 @@ import reactor.core.publisher.Flux;
import reactor.core.subscriber.SubscriberBarrier;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
/**
@@ -47,14 +48,16 @@ public class StringDecoder extends AbstractRawByteStreamDecoder<String> {
public final boolean reduceToSingleBuffer;
private final DataBufferAllocator allocator;
/**
* Create a {@code StringDecoder} that decodes a bytes stream to a String stream
*
* <p>By default, this decoder will buffer bytes and
* emit a single String as a result.
*/
public StringDecoder() {
this(true);
public StringDecoder(DataBufferAllocator allocator) {
this(allocator, true);
}
/**
@@ -63,9 +66,10 @@ public class StringDecoder extends AbstractRawByteStreamDecoder<String> {
* @param reduceToSingleBuffer whether this decoder should buffer all received items
* and decode a single consolidated String or re-emit items as they are provided
*/
public StringDecoder(boolean reduceToSingleBuffer) {
super(new MimeType("text", "plain", DEFAULT_CHARSET));
public StringDecoder(DataBufferAllocator allocator, boolean reduceToSingleBuffer) {
super(allocator, new MimeType("text", "plain", DEFAULT_CHARSET));
this.reduceToSingleBuffer = reduceToSingleBuffer;
this.allocator = allocator;
}
@Override
@@ -75,18 +79,20 @@ public class StringDecoder extends AbstractRawByteStreamDecoder<String> {
}
@Override
public SubscriberBarrier<ByteBuffer, ByteBuffer> subscriberBarrier(Subscriber<? super ByteBuffer> subscriber) {
public SubscriberBarrier<DataBuffer, DataBuffer> subscriberBarrier(
Subscriber<? super DataBuffer> subscriber) {
if (reduceToSingleBuffer) {
return new ReduceSingleByteStreamBarrier(subscriber);
return new ReduceSingleByteStreamBarrier(subscriber, allocator);
}
else {
return new SubscriberBarrier(subscriber);
return new SubscriberBarrier<DataBuffer, DataBuffer>(subscriber);
}
}
@Override
public Flux<String> decodeInternal(Publisher<ByteBuffer> inputStream, ResolvableType type, MimeType mimeType, Object... hints) {
public Flux<String> decodeInternal(Publisher<DataBuffer> inputStream,
ResolvableType type, MimeType mimeType, Object... hints) {
Charset charset;
if (mimeType != null && mimeType.getCharSet() != null) {
charset = mimeType.getCharSet();
@@ -94,7 +100,11 @@ public class StringDecoder extends AbstractRawByteStreamDecoder<String> {
else {
charset = DEFAULT_CHARSET;
}
return Flux.from(inputStream).map(content -> new String(content.duplicate().array(), charset));
return Flux.from(inputStream).map(content -> {
byte[] bytes = new byte[content.readableByteCount()];
content.read(bytes);
return new String(bytes, charset);
});
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@@ -24,6 +23,8 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.MimeType;
/**
@@ -32,13 +33,12 @@ import org.springframework.util.MimeType;
* @author Sebastien Deleuze
* @see StringDecoder
*/
public class StringEncoder extends AbstractEncoder<String> {
public class StringEncoder extends AbstractAllocatingEncoder<String> {
public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
public StringEncoder() {
super(new MimeType("text", "plain", DEFAULT_CHARSET));
public StringEncoder(DataBufferAllocator allocator) {
super(allocator, new MimeType("text", "plain", DEFAULT_CHARSET));
}
@@ -49,7 +49,7 @@ public class StringEncoder extends AbstractEncoder<String> {
}
@Override
public Flux<ByteBuffer> encode(Publisher<? extends String> elementStream,
public Flux<DataBuffer> encode(Publisher<? extends String> elementStream,
ResolvableType type, MimeType mimeType, Object... hints) {
Charset charset;
@@ -59,7 +59,12 @@ public class StringEncoder extends AbstractEncoder<String> {
else {
charset = DEFAULT_CHARSET;
}
return Flux.from(elementStream).map(s -> ByteBuffer.wrap(s.getBytes(charset)));
return Flux.from(elementStream).map(s -> {
byte[] bytes = s.getBytes(charset);
DataBuffer dataBuffer = allocator().allocateBuffer(bytes.length);
dataBuffer.write(bytes);
return dataBuffer;
});
}
}

View File

@@ -0,0 +1,133 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
/**
* Basic abstraction over byte buffers.
*
* @author Arjen Poutsma
*/
public interface DataBuffer {
/**
* Gets the byte at the specified index.
* @param index the index
* @return the byte at the specified index
* @throws IndexOutOfBoundsException if the given index is out of bounds
*/
byte get(int index);
/**
* Returns the number of bytes that can be read from this data buffer.
* @return the readable byte count
*/
int readableByteCount();
/**
* Reads a single byte from the current reading position of this data buffer.
* @return the byte at this buffer's current reading position
*/
byte read();
/**
* Reads this buffer's data into the specified destination, starting at the current
* reading position of this buffer.
*
* @param destination the array into which the bytes are to be written
* @return this buffer
*/
DataBuffer read(byte[] destination);
/**
* Reads at most {@code length} bytes of this buffer into the specified destination,
* starting at the current reading position of this buffer.
* @param destination the array into which the bytes are to be written
* @param offset the index within {@code destination} of the first byte to be written
* @param length the maximum number of bytes to be written in {@code destination}
* @return this buffer
*/
DataBuffer read(byte[] destination, int offset, int length);
/**
* Write a single byte into this buffer at the current writing position.
* @param b the byte to be written
* @return this buffer
*/
DataBuffer write(byte b);
/**
* Writes the given source into this buffer, startin at the current writing position
* of this buffer.
* @param source the bytes to be written into this buffer
* @return this buffer
*/
DataBuffer write(byte[] source);
/**
* Writes at most {@code length} bytes of the given source into this buffer, starting
* at the current writing position of this buffer.
* @param source the bytes to be written into this buffer
* @param offset the index withing {@code source} to start writing from
* @param length the maximum number of bytes to be written from {@code source}
* @return this buffer
*/
DataBuffer write(byte[] source, int offset, int length);
/**
* Writes one or more {@link DataBuffer} to this buffer, starting at the current
* writing position.
* @param buffers the byte buffers to write into this buffer
* @return this buffer
*/
DataBuffer write(DataBuffer... buffers);
/**
* Writes one or more {@link ByteBuffer} to this buffer, starting at the current
* writing position.
* @param buffers the byte buffers to write into this buffer
* @return this buffer
*/
DataBuffer write(ByteBuffer... buffers);
/**
* Exposes this buffer's bytes as a {@link ByteBuffer}. Data between this {@code
* DataBuffer} and the returned {@code ByteBuffer} is shared; though changes in the
* returned buffer's {@linkplain ByteBuffer#position() position} will not be reflected
* in the position(s) of this data buffer.
* @return this data buffer as a byte buffer
*/
ByteBuffer asByteBuffer();
/**
* Exposes this buffer's data as an {@link InputStream}. Both data and position are
* shared between the returned stream and this data buffer.
* @return this data buffer as an input stream
*/
InputStream asInputStream();
/**
* Exposes this buffer's data as an {@link OutputStream}. Both data and position are
* shared between the returned stream and this data buffer.
* @return this data buffer as an output stream
*/
OutputStream asOutputStream();
}

View File

@@ -0,0 +1,52 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.nio.ByteBuffer;
/**
* A factory for {@link DataBuffer}s, allowing for allocation of heap-based and direct
* data buffers.
*
* @author Arjen Poutsma
* @see DataBuffer
*/
public interface DataBufferAllocator {
/**
* Allocates a data buffer of a default initial capacity. Depending on the underlying
* implementation and its configuration, this will be heap-based or direct buffer.
* @return the allocated buffer
*/
DataBuffer allocateBuffer();
/**
* Allocates a data buffer of the given initial capacity. Depending on the underlying
* implementation and its configuration, this will be heap-based or direct buffer.
* @param initialCapacity the initial capacity of the buffer to allocateBuffer
* @return the allocated buffer
*/
DataBuffer allocateBuffer(int initialCapacity);
/**
* Wraps the given {@link ByteBuffer} in a {@code DataBuffer}.
* @param byteBuffer the NIO byte buffer to wrap
* @return the wrapped buffer
*/
DataBuffer wrap(ByteBuffer byteBuffer);
}

View File

@@ -0,0 +1,285 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.function.Function;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
/**
* Default implementation of the {@link DataBuffer} interface that uses a {@link
* ByteBuffer} internally, with separate read and write positions. Constructed
* using the {@link DefaultDataBufferAllocator}.
*
* @author Arjen Poutsma
* @see DefaultDataBufferAllocator
*/
public class DefaultDataBuffer implements DataBuffer {
private ByteBuffer byteBuffer;
private int readPosition;
private int writePosition;
/**
* Creates a new {@code DefaultDataBuffer} based on the given {@code ByteBuffer}. Both
* reading and writing position of this buffer are based on the current {@linkplain
* ByteBuffer#position() position} of the given buffer.
* @param byteBuffer the buffer to base this buffer on
*/
DefaultDataBuffer(ByteBuffer byteBuffer) {
this(byteBuffer, byteBuffer.position(), byteBuffer.position());
}
DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition) {
Assert.notNull(byteBuffer, "'byteBuffer' must not be null");
Assert.isTrue(readPosition >= 0, "'readPosition' must be 0 or higher");
Assert.isTrue(writePosition >= 0, "'writePosition' must be 0 or higher");
Assert.isTrue(readPosition <= writePosition,
"'readPosition' must be smaller than or equal to 'writePosition'");
this.byteBuffer = byteBuffer;
this.readPosition = readPosition;
this.writePosition = writePosition;
}
/**
* Directly exposes the native {@code ByteBuffer} that this buffer is based on.
* @return the wrapped byte buffer
*/
public ByteBuffer getNativeBuffer() {
return this.byteBuffer;
}
@Override
public byte get(int index) {
return this.byteBuffer.get(index);
}
@Override
public int readableByteCount() {
return this.writePosition - this.readPosition;
}
@Override
public byte read() {
return readInternal(ByteBuffer::get);
}
@Override
public DefaultDataBuffer read(byte[] destination) {
Assert.notNull(destination, "'destination' must not be null");
readInternal(b -> b.get(destination));
return this;
}
@Override
public DefaultDataBuffer read(byte[] destination, int offset, int length) {
Assert.notNull(destination, "'destination' must not be null");
readInternal(b -> b.get(destination, offset, length));
return this;
}
/**
* Internal read method that keeps track of the {@link #readPosition} before and after
* applying the given function on {@link #byteBuffer}.
*/
private <T> T readInternal(Function<ByteBuffer, T> function) {
this.byteBuffer.position(this.readPosition);
T result = function.apply(this.byteBuffer);
this.readPosition = this.byteBuffer.position();
return result;
}
@Override
public DefaultDataBuffer write(byte b) {
ensureExtraCapacity(1);
writeInternal(buffer -> buffer.put(b));
return this;
}
@Override
public DefaultDataBuffer write(byte[] source) {
Assert.notNull(source, "'source' must not be null");
ensureExtraCapacity(source.length);
writeInternal(buffer -> buffer.put(source));
return this;
}
@Override
public DefaultDataBuffer write(byte[] source, int offset, int length) {
Assert.notNull(source, "'source' must not be null");
ensureExtraCapacity(length);
writeInternal(buffer -> buffer.put(source, offset, length));
return this;
}
@Override
public DataBuffer write(DataBuffer... buffers) {
if (!ObjectUtils.isEmpty(buffers)) {
ByteBuffer[] byteBuffers =
Arrays.stream(buffers).map(DataBuffer::asByteBuffer)
.toArray(ByteBuffer[]::new);
write(byteBuffers);
}
return this;
}
@Override
public DefaultDataBuffer write(ByteBuffer... byteBuffers) {
Assert.notEmpty(byteBuffers, "'byteBuffers' must not be empty");
int extraCapacity =
Arrays.stream(byteBuffers).mapToInt(ByteBuffer::remaining).sum();
ensureExtraCapacity(extraCapacity);
Arrays.stream(byteBuffers)
.forEach(byteBuffer -> writeInternal(buffer -> buffer.put(byteBuffer)));
return this;
}
/**
* Internal write method that keeps track of the {@link #writePosition} befor eand
* after applying the given function on {@link #byteBuffer}.
*/
private <T> T writeInternal(Function<ByteBuffer, T> function) {
this.byteBuffer.position(this.writePosition);
T result = function.apply(this.byteBuffer);
this.writePosition = this.byteBuffer.position();
return result;
}
@Override
public ByteBuffer asByteBuffer() {
ByteBuffer duplicate = this.byteBuffer.duplicate();
duplicate.position(this.readPosition);
duplicate.limit(this.writePosition);
return duplicate;
}
@Override
public InputStream asInputStream() {
return new DefaultDataBufferInputStream();
}
@Override
public OutputStream asOutputStream() {
return new DefaultDataBufferOutputStream();
}
private void ensureExtraCapacity(int extraCapacity) {
int neededCapacity = this.writePosition + extraCapacity;
if (neededCapacity > this.byteBuffer.capacity()) {
grow(neededCapacity);
}
}
private void grow(int minCapacity) {
ByteBuffer oldBuffer = this.byteBuffer;
ByteBuffer newBuffer =
(oldBuffer.isDirect() ? ByteBuffer.allocateDirect(minCapacity) :
ByteBuffer.allocate(minCapacity));
oldBuffer.position(this.readPosition);
newBuffer.put(oldBuffer);
this.byteBuffer = newBuffer;
oldBuffer.clear();
}
@Override
public int hashCode() {
return this.byteBuffer.hashCode();
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
else if (obj instanceof DefaultDataBuffer) {
DefaultDataBuffer other = (DefaultDataBuffer) obj;
return this.byteBuffer.equals(other.byteBuffer);
}
return false;
}
@Override
public String toString() {
return this.byteBuffer.toString();
}
private class DefaultDataBufferInputStream extends InputStream {
@Override
public int available() throws IOException {
return byteBuffer.limit() - readPosition;
}
@Override
public int read() {
return readInternal(
buffer -> buffer.hasRemaining() ? buffer.get() & 0xFF : -1);
}
@Override
public int read(byte[] bytes, int off, int len) throws IOException {
return readInternal(buffer -> {
if (buffer.hasRemaining()) {
int minLen = Math.min(len, buffer.remaining());
buffer.get(bytes, off, minLen);
return minLen;
}
else {
return -1;
}
});
}
}
private class DefaultDataBufferOutputStream extends OutputStream {
@Override
public void write(int b) throws IOException {
ensureExtraCapacity(1);
writeInternal(buffer -> buffer.put((byte) b));
}
@Override
public void write(byte[] bytes, int off, int len) throws IOException {
ensureExtraCapacity(len);
writeInternal(buffer -> buffer.put(bytes, off, len));
}
}
}

View File

@@ -0,0 +1,73 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.nio.ByteBuffer;
/**
* Default implementation of the {@code DataBufferAllocator} interface.
*
* @author Arjen Poutsma
*/
public class DefaultDataBufferAllocator implements DataBufferAllocator {
public static final int DEFAULT_INITIAL_CAPACITY = 256;
private final boolean preferDirect;
/**
* Creates a new {@code DefaultDataBufferAllocator} with default settings.
*/
public DefaultDataBufferAllocator() {
this(false);
}
/**
* Creates a new {@code DefaultDataBufferAllocator}, indicating whether direct buffers
* should be created by {@link #allocateBuffer(int)}.
* @param preferDirect {@code true} if direct buffers are to be preferred; {@code
* false} otherwise
*/
public DefaultDataBufferAllocator(boolean preferDirect) {
this.preferDirect = preferDirect;
}
@Override
public DataBuffer allocateBuffer() {
return allocateBuffer(DEFAULT_INITIAL_CAPACITY);
}
@Override
public DefaultDataBuffer allocateBuffer(int initialCapacity) {
return this.preferDirect ?
new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity)) :
new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity));
}
@Override
public DataBuffer wrap(ByteBuffer byteBuffer) {
ByteBuffer sliced = byteBuffer.slice();
return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining());
}
@Override
public String toString() {
return "DefaultDataBufferFactory - preferDirect: " + this.preferDirect;
}
}

View File

@@ -0,0 +1,196 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
/**
* Implementation of the {@code DataBuffer} interface that wraps a Netty {@link ByteBuf}.
* Typically constructed using the {@link NettyDataBufferAllocator}.
*
* @author Arjen Poutsma
*/
public class NettyDataBuffer implements DataBuffer {
private ByteBuf byteBuf;
/**
* Creates a new {@code NettyDataBuffer} based on the given {@code ByteBuff}.
* @param byteBuf the buffer to base this buffer on
*/
public NettyDataBuffer(ByteBuf byteBuf) {
Assert.notNull(byteBuf, "'byteBuf' must not be null");
this.byteBuf = byteBuf;
}
/**
* Directly exposes the native {@code ByteBuf} that this buffer is based on.
* @return the wrapped byte buffer
*/
public ByteBuf getNativeBuffer() {
return this.byteBuf;
}
@Override
public byte get(int index) {
return this.byteBuf.getByte(index);
}
@Override
public int readableByteCount() {
return this.byteBuf.readableBytes();
}
@Override
public byte read() {
return this.byteBuf.readByte();
}
@Override
public NettyDataBuffer read(byte[] destination) {
this.byteBuf.readBytes(destination);
return this;
}
@Override
public NettyDataBuffer read(byte[] destination, int offset, int length) {
this.byteBuf.readBytes(destination, offset, length);
return this;
}
@Override
public NettyDataBuffer write(byte b) {
this.byteBuf.writeByte(b);
return this;
}
@Override
public NettyDataBuffer write(byte[] source) {
this.byteBuf.writeBytes(source);
return this;
}
@Override
public NettyDataBuffer write(byte[] source, int offset, int length) {
this.byteBuf.writeBytes(source, offset, length);
return this;
}
@Override
public NettyDataBuffer write(DataBuffer... buffers) {
if (!ObjectUtils.isEmpty(buffers)) {
if (buffers[0] instanceof NettyDataBuffer) {
ByteBuf[] nativeBuffers = Arrays.stream(buffers)
.map(b -> ((NettyDataBuffer) b).getNativeBuffer())
.toArray(ByteBuf[]::new);
write(nativeBuffers);
}
else {
ByteBuffer[] byteBuffers =
Arrays.stream(buffers).map(DataBuffer::asByteBuffer)
.toArray(ByteBuffer[]::new);
write(byteBuffers);
}
}
return this;
}
@Override
public NettyDataBuffer write(ByteBuffer... buffers) {
Assert.notNull(buffers, "'buffers' must not be null");
ByteBuf[] wrappedBuffers = Arrays.stream(buffers).map(Unpooled::wrappedBuffer)
.toArray(ByteBuf[]::new);
return write(wrappedBuffers);
}
/**
* Writes one or more Netty {@link ByteBuf}s to this buffer, starting at the current
* writing position.
* @param byteBufs the buffers to write into this buffer
* @return this buffer
*/
public NettyDataBuffer write(ByteBuf... byteBufs) {
Assert.notNull(byteBufs, "'byteBufs' must not be null");
CompositeByteBuf composite =
new CompositeByteBuf(this.byteBuf.alloc(), this.byteBuf.isDirect(),
byteBufs.length + 1);
composite.addComponent(this.byteBuf);
composite.addComponents(byteBufs);
int writerIndex = this.byteBuf.readableBytes() +
Arrays.stream(byteBufs).mapToInt(ByteBuf::readableBytes).sum();
composite.writerIndex(writerIndex);
this.byteBuf = composite;
return this;
}
@Override
public ByteBuffer asByteBuffer() {
return this.byteBuf.nioBuffer();
}
@Override
public InputStream asInputStream() {
return new ByteBufInputStream(this.byteBuf);
}
@Override
public OutputStream asOutputStream() {
return new ByteBufOutputStream(this.byteBuf);
}
@Override
public int hashCode() {
return this.byteBuf.hashCode();
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
else if (obj instanceof NettyDataBuffer) {
NettyDataBuffer other = (NettyDataBuffer) obj;
return this.byteBuf.equals(other.byteBuf);
}
return false;
}
@Override
public String toString() {
return this.byteBuf.toString();
}
}

View File

@@ -0,0 +1,82 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.nio.ByteBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import org.springframework.util.Assert;
/**
* Implementation of the {@code DataBufferAllocator} interface based on a Netty
* {@link ByteBufAllocator}.
*
* @author Arjen Poutsma
* @see io.netty.buffer.PooledByteBufAllocator
* @see io.netty.buffer.UnpooledByteBufAllocator
*/
public class NettyDataBufferAllocator implements DataBufferAllocator {
private final ByteBufAllocator byteBufAllocator;
/**
* Creates a new {@code NettyDataBufferAllocator} based on the given allocator.
* @param byteBufAllocator the allocator to use
* @see io.netty.buffer.PooledByteBufAllocator
* @see io.netty.buffer.UnpooledByteBufAllocator
*/
public NettyDataBufferAllocator(ByteBufAllocator byteBufAllocator) {
Assert.notNull(byteBufAllocator, "'byteBufAllocator' must not be null");
this.byteBufAllocator = byteBufAllocator;
}
@Override
public NettyDataBuffer allocateBuffer() {
ByteBuf byteBuf = this.byteBufAllocator.buffer();
return new NettyDataBuffer(byteBuf);
}
@Override
public NettyDataBuffer allocateBuffer(int initialCapacity) {
ByteBuf byteBuf = this.byteBufAllocator.buffer(initialCapacity);
return new NettyDataBuffer(byteBuf);
}
@Override
public NettyDataBuffer wrap(ByteBuffer byteBuffer) {
ByteBuf byteBuf = Unpooled.wrappedBuffer(byteBuffer);
return new NettyDataBuffer(byteBuf);
}
/**
* Wraps the given Netty {@link ByteBuf} in a {@code NettyDataBuffer}.
* @param byteBuf the Netty byte buffer to wrap
* @return the wrapped buffer
*/
public NettyDataBuffer wrap(ByteBuf byteBuf) {
return new NettyDataBuffer(byteBuf);
}
@Override
public String toString() {
return "NettyDataBufferAllocator (" + this.byteBufAllocator + ")";
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,60 +14,56 @@
* limitations under the License.
*/
package org.springframework.util;
package org.springframework.core.io.buffer.support;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.rx.Stream;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.Assert;
/**
* {@code InputStream} implementation based on a byte array {@link Publisher}.
*
* @author Arjen Poutsma
* @author Sebastien Deleuze
* @author Stephane Maldini
*/
public class ByteBufferPublisherInputStream extends InputStream {
class DataBufferPublisherInputStream extends InputStream {
private final BlockingQueue<ByteBuffer> queue;
private final AtomicBoolean completed = new AtomicBoolean();
private ByteBufferInputStream currentStream;
private boolean completed;
private final BlockingQueue<DataBuffer> queue;
private InputStream currentStream;
/**
* Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher.
*
* @param publisher the publisher to use
*/
public ByteBufferPublisherInputStream(Publisher<ByteBuffer> publisher) {
public DataBufferPublisherInputStream(Publisher<DataBuffer> publisher) {
this(publisher, 1);
}
/**
* Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher.
*
* @param publisher the publisher to use
* @param publisher the publisher to use
* @param requestSize the {@linkplain Subscription#request(long) request size} to use
* on the publisher bound to Integer MAX
*/
public ByteBufferPublisherInputStream(Publisher<ByteBuffer> publisher, int requestSize) {
public DataBufferPublisherInputStream(Publisher<DataBuffer> publisher,
int requestSize) {
Assert.notNull(publisher, "'publisher' must not be null");
// TODO Avoid using Reactor Stream, it should not be a mandatory dependency of Spring Reactive
this.queue = Stream.from(publisher).toBlockingQueue(requestSize);
}
@Override
public int available() throws IOException {
if (completed) {
if (completed.get()) {
return 0;
}
InputStream is = currentStream();
@@ -76,7 +72,7 @@ public class ByteBufferPublisherInputStream extends InputStream {
@Override
public int read() throws IOException {
if (completed) {
if (completed.get()) {
return -1;
}
InputStream is = currentStream();
@@ -94,7 +90,7 @@ public class ByteBufferPublisherInputStream extends InputStream {
@Override
public int read(byte[] b, int off, int len) throws IOException {
if (completed) {
if (completed.get()) {
return -1;
}
InputStream is = currentStream();
@@ -128,26 +124,28 @@ public class ByteBufferPublisherInputStream extends InputStream {
try {
if (this.currentStream != null && this.currentStream.available() > 0) {
return this.currentStream;
} else {
}
else {
// take() blocks until next or complete() then return null,
// but that's OK since this is a *blocking* InputStream
ByteBuffer signal = this.queue.take();
if(signal == null){
this.completed = true;
DataBuffer signal = this.queue.take();
if (signal == null) {
this.completed.set(true);
return null;
}
this.currentStream = new ByteBufferInputStream(signal);
this.currentStream = signal.asInputStream();
return this.currentStream;
}
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
catch (Throwable error ){
this.completed = true;
catch (Throwable error) {
this.completed.set(true);
throw new IOException(error);
}
throw new IOException();
}
}

View File

@@ -0,0 +1,49 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer.support;
import java.io.InputStream;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
*/
public abstract class DataBufferUtils {
public static Flux<Byte> toPublisher(DataBuffer buffer) {
Assert.notNull(buffer, "'buffer' must not be null");
byte[] bytes1 = new byte[buffer.readableByteCount()];
buffer.read(bytes1);
Byte[] bytes2 = new Byte[bytes1.length];
for (int i = 0; i < bytes1.length; i++) {
bytes2[i] = bytes1[i];
}
return Flux.fromArray(bytes2);
}
public static InputStream toInputStream(Publisher<DataBuffer> publisher) {
return new DataBufferPublisherInputStream(publisher);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@
package org.springframework.http;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
/**
* An "reactive" HTTP input message that exposes the input as {@link Publisher}.
*
@@ -35,6 +35,6 @@ public interface ReactiveHttpInputMessage extends HttpMessage {
* Return the body of the message as a {@link Publisher}.
* @return the body content publisher
*/
Flux<ByteBuffer> getBody();
Flux<DataBuffer> getBody();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@
package org.springframework.http;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
/**
* A "reactive" HTTP output message that accepts output as a {@link Publisher}.
*
@@ -38,6 +38,6 @@ public interface ReactiveHttpOutputMessage extends HttpMessage {
* @param body the body content publisher
* @return a publisher that indicates completion or error.
*/
Mono<Void> setBody(Publisher<ByteBuffer> body);
Mono<Void> setBody(Publisher<DataBuffer> body);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -15,12 +15,11 @@
*/
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
/**
@@ -46,7 +45,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
}
@Override
public Mono<Void> setBody(Publisher<ByteBuffer> publisher) {
public Mono<Void> setBody(Publisher<DataBuffer> publisher) {
return Flux.from(publisher).lift(new WriteWithOperator<>(writeWithPublisher -> {
writeHeaders();
return setBodyInternal(writeWithPublisher);
@@ -57,7 +56,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
* Implement this method to write to the underlying the response.
* @param publisher the publisher to write with
*/
protected abstract Mono<Void> setBodyInternal(Publisher<ByteBuffer> publisher);
protected abstract Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher);
@Override
public void writeHeaders() {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import reactor.io.buffer.Buffer;
import reactor.io.net.ReactiveChannelHandler;
import reactor.io.net.http.HttpChannel;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.Assert;
/**
@@ -30,15 +31,19 @@ public class ReactorHttpHandlerAdapter
private final HttpHandler httpHandler;
private final DataBufferAllocator allocator;
public ReactorHttpHandlerAdapter(HttpHandler httpHandler) {
public ReactorHttpHandlerAdapter(HttpHandler httpHandler,
DataBufferAllocator allocator) {
Assert.notNull(httpHandler, "'httpHandler' is required.");
this.httpHandler = httpHandler;
this.allocator = allocator;
}
@Override
public Mono<Void> apply(HttpChannel<Buffer, Buffer> channel) {
ReactorServerHttpRequest adaptedRequest = new ReactorServerHttpRequest(channel);
ReactorServerHttpRequest adaptedRequest =
new ReactorServerHttpRequest(channel, allocator);
ReactorServerHttpResponse adaptedResponse = new ReactorServerHttpResponse(channel);
return this.httpHandler.handle(adaptedRequest, adaptedResponse);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -27,6 +27,8 @@ import reactor.io.buffer.Buffer;
import reactor.io.net.http.HttpChannel;
import reactor.io.net.http.model.Cookie;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -41,10 +43,14 @@ public class ReactorServerHttpRequest extends AbstractServerHttpRequest {
private final HttpChannel<Buffer, ?> channel;
private final DataBufferAllocator allocator;
public ReactorServerHttpRequest(HttpChannel<Buffer, ?> request) {
Assert.notNull("'request' must not be null.");
public ReactorServerHttpRequest(HttpChannel<Buffer, ?> request,
DataBufferAllocator allocator) {
Assert.notNull("'request' must not be null");
Assert.notNull(allocator, "'allocator' must not be null");
this.channel = request;
this.allocator = allocator;
}
@@ -84,8 +90,11 @@ public class ReactorServerHttpRequest extends AbstractServerHttpRequest {
}
@Override
public Flux<ByteBuffer> getBody() {
return Flux.from(this.channel.input()).map(Buffer::byteBuffer);
public Flux<DataBuffer> getBody() {
return Flux.from(this.channel.input()).map(bytes -> {
ByteBuffer byteBuffer = bytes.byteBuffer();
return allocator.wrap(byteBuffer);
});
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -15,8 +15,6 @@
*/
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -25,6 +23,7 @@ import reactor.io.net.http.HttpChannel;
import reactor.io.net.http.model.Cookie;
import reactor.io.net.http.model.Status;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpStatus;
import org.springframework.util.Assert;
@@ -56,8 +55,9 @@ public class ReactorServerHttpResponse extends AbstractServerHttpResponse {
}
@Override
protected Mono<Void> setBodyInternal(Publisher<ByteBuffer> publisher) {
return Mono.from(this.channel.writeWith(Flux.from(publisher).map(Buffer::new)));
protected Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher) {
return Mono.from(this.channel.writeWith(
Flux.from(publisher).map(buffer -> new Buffer(buffer.asByteBuffer()))));
}
@Override

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import org.reactivestreams.Publisher;
import reactor.core.converter.RxJava1ObservableConverter;
import rx.Observable;
import org.springframework.core.io.buffer.NettyDataBufferAllocator;
import org.springframework.util.Assert;
/**
@@ -33,15 +34,20 @@ public class RxNettyHttpHandlerAdapter implements RequestHandler<ByteBuf, ByteBu
private final HttpHandler httpHandler;
private final NettyDataBufferAllocator allocator;
public RxNettyHttpHandlerAdapter(HttpHandler httpHandler) {
Assert.notNull(httpHandler, "'httpHandler' is required.");
public RxNettyHttpHandlerAdapter(HttpHandler httpHandler,
NettyDataBufferAllocator allocator) {
Assert.notNull(httpHandler, "'httpHandler' is required");
Assert.notNull(allocator, "'allocator' must not be null");
this.httpHandler = httpHandler;
this.allocator = allocator;
}
@Override
public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
RxNettyServerHttpRequest adaptedRequest = new RxNettyServerHttpRequest(request);
RxNettyServerHttpRequest adaptedRequest =
new RxNettyServerHttpRequest(request, allocator);
RxNettyServerHttpResponse adaptedResponse = new RxNettyServerHttpResponse(response);
Publisher<Void> result = this.httpHandler.handle(adaptedRequest, adaptedResponse);
return RxJava1ObservableConverter.from(result);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@ package org.springframework.http.server.reactive;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -30,6 +29,8 @@ import reactor.core.converter.RxJava1ObservableConverter;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.NettyDataBufferAllocator;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -45,9 +46,13 @@ public class RxNettyServerHttpRequest extends AbstractServerHttpRequest {
private final HttpServerRequest<ByteBuf> request;
private final NettyDataBufferAllocator allocator;
public RxNettyServerHttpRequest(HttpServerRequest<ByteBuf> request) {
Assert.notNull("'request', request must not be null.");
public RxNettyServerHttpRequest(HttpServerRequest<ByteBuf> request,
NettyDataBufferAllocator allocator) {
Assert.notNull("'request', request must not be null");
Assert.notNull(allocator, "'allocator' must not be null");
this.allocator = allocator;
this.request = request;
}
@@ -88,8 +93,8 @@ public class RxNettyServerHttpRequest extends AbstractServerHttpRequest {
}
@Override
public Flux<ByteBuffer> getBody() {
Observable<ByteBuffer> content = this.request.getContent().map(ByteBuf::nioBuffer);
public Flux<DataBuffer> getBody() {
Observable<DataBuffer> content = this.request.getContent().map(allocator::wrap);
content = content.concatWith(Observable.empty()); // See GH issue #58
return RxJava1ObservableConverter.from(content);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.codec.http.cookie.DefaultCookie;
@@ -27,6 +25,7 @@ import reactor.core.converter.RxJava1ObservableConverter;
import reactor.core.publisher.Mono;
import rx.Observable;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpStatus;
import org.springframework.util.Assert;
@@ -58,15 +57,15 @@ public class RxNettyServerHttpResponse extends AbstractServerHttpResponse {
}
@Override
protected Mono<Void> setBodyInternal(Publisher<ByteBuffer> publisher) {
protected Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher) {
Observable<byte[]> content = RxJava1ObservableConverter.from(publisher).map(this::toBytes);
Observable<Void> completion = this.response.writeBytes(content);
return RxJava1ObservableConverter.from(completion).after();
}
private byte[] toBytes(ByteBuffer buffer) {
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
private byte[] toBytes(DataBuffer buffer) {
byte[] bytes = new byte[buffer.readableByteCount()];
buffer.read(bytes);
return bytes;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,8 +17,6 @@
package org.springframework.http.server.reactive;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicLong;
import javax.servlet.AsyncContext;
import javax.servlet.ReadListener;
@@ -38,6 +36,9 @@ import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.HttpStatus;
import org.springframework.util.Assert;
@@ -55,11 +56,16 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
private HttpHandler handler;
private DataBufferAllocator allocator = new DefaultDataBufferAllocator();
public void setHandler(HttpHandler handler) {
this.handler = handler;
}
public void setAllocator(DataBufferAllocator allocator) {
this.allocator = allocator;
}
@Override
protected void service(HttpServletRequest servletRequest, HttpServletResponse servletResponse)
@@ -68,11 +74,13 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
AsyncContext context = servletRequest.startAsync();
ServletAsyncContextSynchronizer synchronizer = new ServletAsyncContextSynchronizer(context);
RequestBodyPublisher requestBody = new RequestBodyPublisher(synchronizer, BUFFER_SIZE);
RequestBodyPublisher requestBody =
new RequestBodyPublisher(synchronizer, allocator, BUFFER_SIZE);
ServletServerHttpRequest request = new ServletServerHttpRequest(servletRequest, requestBody);
servletRequest.getInputStream().setReadListener(requestBody);
ResponseBodySubscriber responseBodySubscriber = new ResponseBodySubscriber(synchronizer);
ResponseBodySubscriber responseBodySubscriber =
new ResponseBodySubscriber(synchronizer, allocator);
ServletServerHttpResponse response = new ServletServerHttpResponse(servletResponse,
publisher -> Mono.from(subscriber -> publisher.subscribe(responseBodySubscriber)));
servletResponse.getOutputStream().setWriteListener(responseBodySubscriber);
@@ -81,30 +89,32 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
this.handler.handle(request, response).subscribe(resultSubscriber);
}
private static class RequestBodyPublisher implements ReadListener, Publisher<ByteBuffer> {
private static class RequestBodyPublisher
implements ReadListener, Publisher<DataBuffer> {
private final ServletAsyncContextSynchronizer synchronizer;
private final DataBufferAllocator allocator;
private final byte[] buffer;
private final DemandCounter demand = new DemandCounter();
private Subscriber<? super ByteBuffer> subscriber;
private Subscriber<? super DataBuffer> subscriber;
private boolean stalled;
private boolean cancelled;
public RequestBodyPublisher(ServletAsyncContextSynchronizer synchronizer, int bufferSize) {
public RequestBodyPublisher(ServletAsyncContextSynchronizer synchronizer,
DataBufferAllocator allocator, int bufferSize) {
this.synchronizer = synchronizer;
this.allocator = allocator;
this.buffer = new byte[bufferSize];
}
@Override
public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
public void subscribe(Subscriber<? super DataBuffer> subscriber) {
if (subscriber == null) {
throw new NullPointerException();
}
@@ -146,11 +156,11 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
}
else if (read > 0) {
this.demand.decrement();
byte[] copy = Arrays.copyOf(this.buffer, read);
// logger.debug("Next: " + new String(copy, UTF_8));
DataBuffer dataBuffer = allocator.allocateBuffer(read);
dataBuffer.write(this.buffer, 0, read);
this.subscriber.onNext(ByteBuffer.wrap(copy));
this.subscriber.onNext(dataBuffer);
}
}
@@ -265,19 +275,23 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
}
}
private static class ResponseBodySubscriber implements WriteListener, Subscriber<ByteBuffer> {
private static class ResponseBodySubscriber
implements WriteListener, Subscriber<DataBuffer> {
private final ServletAsyncContextSynchronizer synchronizer;
private final DataBufferAllocator allocator;
private Subscription subscription;
private ByteBuffer buffer;
private DataBuffer buffer;
private volatile boolean subscriberComplete = false;
public ResponseBodySubscriber(ServletAsyncContextSynchronizer synchronizer) {
public ResponseBodySubscriber(ServletAsyncContextSynchronizer synchronizer,
DataBufferAllocator allocator) {
this.synchronizer = synchronizer;
this.allocator = allocator;
}
@@ -288,8 +302,7 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
}
@Override
public void onNext(ByteBuffer bytes) {
public void onNext(DataBuffer bytes) {
Assert.isNull(buffer);
this.buffer = bytes;
@@ -321,8 +334,8 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
if (ready) {
if (this.buffer != null) {
byte[] bytes = new byte[this.buffer.remaining()];
this.buffer.get(bytes);
byte[] bytes = new byte[this.buffer.readableByteCount()];
this.buffer.read(bytes);
this.buffer = null;
output.write(bytes);
if (!subscriberComplete) {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@ package org.springframework.http.server.reactive;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Enumeration;
@@ -30,6 +29,7 @@ import javax.servlet.http.HttpServletRequest;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -47,10 +47,10 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
private final HttpServletRequest request;
private final Flux<ByteBuffer> requestBodyPublisher;
private final Flux<DataBuffer> requestBodyPublisher;
public ServletServerHttpRequest(HttpServletRequest request, Publisher<ByteBuffer> body) {
public ServletServerHttpRequest(HttpServletRequest request,
Publisher<DataBuffer> body) {
Assert.notNull(request, "'request' must not be null.");
Assert.notNull(body, "'body' must not be null.");
this.request = request;
@@ -125,7 +125,7 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
}
@Override
public Flux<ByteBuffer> getBody() {
public Flux<DataBuffer> getBody() {
return this.requestBodyPublisher;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
@@ -27,6 +26,7 @@ import javax.servlet.http.HttpServletResponse;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
@@ -41,11 +41,11 @@ public class ServletServerHttpResponse extends AbstractServerHttpResponse {
private final HttpServletResponse response;
private final Function<Publisher<ByteBuffer>, Mono<Void>> responseBodyWriter;
private final Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter;
public ServletServerHttpResponse(HttpServletResponse response,
Function<Publisher<ByteBuffer>, Mono<Void>> responseBodyWriter) {
Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter) {
Assert.notNull(response, "'response' must not be null");
Assert.notNull(responseBodyWriter, "'responseBodyWriter' must not be null");
@@ -64,7 +64,7 @@ public class ServletServerHttpResponse extends AbstractServerHttpResponse {
}
@Override
protected Mono<Void> setBodyInternal(Publisher<ByteBuffer> publisher) {
protected Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher) {
return this.responseBodyWriter.apply(publisher);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -33,6 +33,8 @@ import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.xnio.ChannelListener;
import org.xnio.ChannelListeners;
import org.xnio.IoUtils;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.channels.StreamSourceChannel;
import reactor.core.publisher.Mono;
@@ -40,13 +42,10 @@ import reactor.core.subscriber.BaseSubscriber;
import reactor.core.util.BackpressureUtils;
import reactor.core.util.Exceptions;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.util.Assert;
import static org.xnio.ChannelListeners.closingChannelExceptionHandler;
import static org.xnio.ChannelListeners.flushingChannelListener;
import static org.xnio.IoUtils.safeClose;
/**
* @author Marek Hawrylczak
* @author Rossen Stoyanchev
@@ -58,17 +57,21 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
private final HttpHandler delegate;
private final DataBufferAllocator allocator;
public UndertowHttpHandlerAdapter(HttpHandler delegate) {
Assert.notNull(delegate, "'delegate' is required.");
public UndertowHttpHandlerAdapter(HttpHandler delegate,
DataBufferAllocator allocator) {
Assert.notNull(delegate, "'delegate' is required");
Assert.notNull(allocator, "'allocator' must not be null");
this.delegate = delegate;
this.allocator = allocator;
}
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
RequestBodyPublisher requestBody = new RequestBodyPublisher(exchange);
RequestBodyPublisher requestBody = new RequestBodyPublisher(exchange, allocator);
ServerHttpRequest request = new UndertowServerHttpRequest(exchange, requestBody);
ResponseBodySubscriber responseBodySubscriber = new ResponseBodySubscriber(exchange);
@@ -107,8 +110,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
});
}
private static class RequestBodyPublisher implements Publisher<ByteBuffer> {
private static class RequestBodyPublisher implements Publisher<DataBuffer> {
private static final AtomicLongFieldUpdater<RequestBodySubscription> DEMAND =
AtomicLongFieldUpdater.newUpdater(RequestBodySubscription.class, "demand");
@@ -116,16 +118,18 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
private final HttpServerExchange exchange;
private Subscriber<? super ByteBuffer> subscriber;
private final DataBufferAllocator allocator;
private Subscriber<? super DataBuffer> subscriber;
public RequestBodyPublisher(HttpServerExchange exchange) {
public RequestBodyPublisher(HttpServerExchange exchange,
DataBufferAllocator allocator) {
this.exchange = exchange;
this.allocator = allocator;
}
@Override
public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
public void subscribe(Subscriber<? super DataBuffer> subscriber) {
if (subscriber == null) {
throw Exceptions.spec_2_13_exception();
}
@@ -175,11 +179,11 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
private void close() {
if (this.pooledBuffer != null) {
safeClose(this.pooledBuffer);
IoUtils.safeClose(this.pooledBuffer);
this.pooledBuffer = null;
}
if (this.channel != null) {
safeClose(this.channel);
IoUtils.safeClose(this.channel);
this.channel = null;
}
}
@@ -251,7 +255,8 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
private void doOnNext(ByteBuffer buffer) {
this.draining = false;
buffer.flip();
subscriber.onNext(buffer);
DataBuffer dataBuffer = allocator.wrap(buffer);
subscriber.onNext(dataBuffer);
}
private void doOnComplete() {
@@ -315,7 +320,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
}
}
private static class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
private static class ResponseBodySubscriber extends BaseSubscriber<DataBuffer>
implements ChannelListener<StreamSinkChannel> {
private final HttpServerExchange exchange;
@@ -343,8 +348,10 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
}
@Override
public void onNext(ByteBuffer buffer) {
super.onNext(buffer);
public void onNext(DataBuffer dataBuffer) {
super.onNext(dataBuffer);
ByteBuffer buffer = dataBuffer.asByteBuffer();
if (this.responseChannel == null) {
this.responseChannel = exchange.getResponseChannel();
@@ -407,7 +414,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
} while (buffer.hasRemaining() && c > 0);
if (!buffer.hasRemaining()) {
safeClose(this.buffers.remove());
IoUtils.safeClose(this.buffers.remove());
}
} while (!this.buffers.isEmpty() && c > 0);
@@ -461,8 +468,10 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
this.responseChannel.shutdownWrites();
if (!this.responseChannel.flush()) {
this.responseChannel.getWriteSetter().set(flushingChannelListener(
o -> safeClose(this.responseChannel), closingChannelExceptionHandler()));
this.responseChannel.getWriteSetter().set(ChannelListeners
.flushingChannelListener(
o -> IoUtils.safeClose(this.responseChannel),
ChannelListeners.closingChannelExceptionHandler()));
this.responseChannel.resumeWrites();
}
this.responseChannel = null;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@ package org.springframework.http.server.reactive;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -29,6 +28,7 @@ import io.undertow.util.HeaderValues;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -44,10 +44,10 @@ public class UndertowServerHttpRequest extends AbstractServerHttpRequest {
private final HttpServerExchange exchange;
private final Flux<ByteBuffer> body;
private final Flux<DataBuffer> body;
public UndertowServerHttpRequest(HttpServerExchange exchange, Publisher<ByteBuffer> body) {
public UndertowServerHttpRequest(HttpServerExchange exchange,
Publisher<DataBuffer> body) {
Assert.notNull(exchange, "'exchange' is required.");
Assert.notNull(exchange, "'body' is required.");
this.exchange = exchange;
@@ -92,7 +92,7 @@ public class UndertowServerHttpRequest extends AbstractServerHttpRequest {
}
@Override
public Flux<ByteBuffer> getBody() {
public Flux<DataBuffer> getBody() {
return this.body;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
@@ -28,6 +27,7 @@ import io.undertow.util.HttpString;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpStatus;
import org.springframework.util.Assert;
@@ -42,11 +42,11 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse {
private final HttpServerExchange exchange;
private final Function<Publisher<ByteBuffer>, Mono<Void>> responseBodyWriter;
private final Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter;
public UndertowServerHttpResponse(HttpServerExchange exchange,
Function<Publisher<ByteBuffer>, Mono<Void>> responseBodyWriter) {
Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter) {
Assert.notNull(exchange, "'exchange' is required.");
Assert.notNull(responseBodyWriter, "'responseBodyWriter' must not be null");
@@ -66,7 +66,7 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse {
}
@Override
protected Mono<Void> setBodyInternal(Publisher<ByteBuffer> publisher) {
protected Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher) {
return this.responseBodyWriter.apply(publisher);
}

View File

@@ -1,11 +1,11 @@
/*
* Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -21,7 +21,8 @@ import reactor.core.state.Completable;
import reactor.io.buffer.Buffer;
import reactor.io.net.ReactiveNet;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.util.Assert;
@@ -29,29 +30,35 @@ import org.springframework.util.Assert;
* @author Stephane Maldini
*/
public class ReactorHttpServer extends HttpServerSupport
implements InitializingBean, HttpServer, Connectable, Completable {
implements HttpServer, Connectable, Completable {
private ReactorHttpHandlerAdapter reactorHandler;
private reactor.io.net.http.HttpServer<Buffer, Buffer> reactorServer;
private DataBufferAllocator allocator = new DefaultDataBufferAllocator();
private boolean running;
@Override
public boolean isRunning() {
return this.running;
public void setAllocator(DataBufferAllocator allocator) {
this.allocator = allocator;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(getHttpHandler());
this.reactorHandler = new ReactorHttpHandlerAdapter(getHttpHandler());
this.reactorHandler = new ReactorHttpHandlerAdapter(getHttpHandler(), allocator);
this.reactorServer = (getPort() != -1 ? ReactiveNet.httpServer(getPort()) :
ReactiveNet.httpServer());
}
@Override
public boolean isRunning() {
return this.running;
}
@Override
public Object connectedInput() {
return reactorServer;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,8 +17,10 @@
package org.springframework.http.server.reactive.boot;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.buffer.NettyDataBufferAllocator;
import org.springframework.http.server.reactive.RxNettyHttpHandlerAdapter;
import org.springframework.util.Assert;
@@ -26,14 +28,34 @@ import org.springframework.util.Assert;
/**
* @author Rossen Stoyanchev
*/
public class RxNettyHttpServer extends HttpServerSupport implements InitializingBean, HttpServer {
public class RxNettyHttpServer extends HttpServerSupport implements HttpServer {
private RxNettyHttpHandlerAdapter rxNettyHandler;
private io.reactivex.netty.protocol.http.server.HttpServer<ByteBuf, ByteBuf> rxNettyServer;
private NettyDataBufferAllocator allocator;
private boolean running;
public void setAllocator(ByteBufAllocator allocator) {
Assert.notNull(allocator, "'allocator' must not be null");
this.allocator = new NettyDataBufferAllocator(allocator);
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(getHttpHandler());
if (allocator == null) {
allocator = new NettyDataBufferAllocator(UnpooledByteBufAllocator.DEFAULT);
}
this.rxNettyHandler = new RxNettyHttpHandlerAdapter(getHttpHandler(), allocator);
this.rxNettyServer = (getPort() != -1 ?
io.reactivex.netty.protocol.http.server.HttpServer.newServer(getPort()) :
io.reactivex.netty.protocol.http.server.HttpServer.newServer());
}
@Override
public boolean isRunning() {
@@ -41,18 +63,6 @@ public class RxNettyHttpServer extends HttpServerSupport implements Initializing
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(getHttpHandler());
this.rxNettyHandler = new RxNettyHttpHandlerAdapter(getHttpHandler());
this.rxNettyServer = (getPort() != -1 ?
io.reactivex.netty.protocol.http.server.HttpServer.newServer(getPort()) :
io.reactivex.netty.protocol.http.server.HttpServer.newServer());
}
@Override
public void start() {
if (!this.running) {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,24 +19,30 @@ package org.springframework.http.server.reactive.boot;
import io.undertow.Undertow;
import io.undertow.server.HttpHandler;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.server.reactive.UndertowHttpHandlerAdapter;
import org.springframework.util.Assert;
/**
* @author Marek Hawrylczak
*/
public class UndertowHttpServer extends HttpServerSupport implements InitializingBean, HttpServer {
public class UndertowHttpServer extends HttpServerSupport implements HttpServer {
private Undertow server;
private DataBufferAllocator allocator = new DefaultDataBufferAllocator();
private boolean running;
public void setAllocator(DataBufferAllocator allocator) {
this.allocator = allocator;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(getHttpHandler());
HttpHandler handler = new UndertowHttpHandlerAdapter(getHttpHandler());
HttpHandler handler = new UndertowHttpHandlerAdapter(getHttpHandler(), allocator);
int port = (getPort() != -1 ? getPort() : 8080);
this.server = Undertow.builder().addHttpListener(port, "localhost")
.setHandler(handler).build();

View File

@@ -1,53 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.util;
import java.io.IOException;
import java.io.OutputStream;
import reactor.io.buffer.Buffer;
/**
* Simple extension of {@link OutputStream} that uses {@link Buffer} to stream
* the content
*
* @author Sebastien Deleuze
*/
public class BufferOutputStream extends OutputStream {
private Buffer buffer;
public BufferOutputStream(Buffer buffer) {
this.buffer = buffer;
}
@Override
public void write(int b) throws IOException {
buffer.append(b);
}
@Override
public void write(byte[] bytes, int off, int len)
throws IOException {
buffer.append(bytes, off, len);
}
public Buffer getBuffer() {
return buffer;
}
}

View File

@@ -1,55 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.util;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
/**
* Simple {@link InputStream} implementation that exposes currently
* available content of a {@link ByteBuffer}.
*
* From Jackson <a href="https://github.com/FasterXML/jackson-databind/blob/master/src/main/java/com/fasterxml/jackson/databind/util/ByteBufferBackedInputStream.java">ByteBufferBackedInputStream</a>
*/
public class ByteBufferInputStream extends InputStream {
protected final ByteBuffer b;
public ByteBufferInputStream(ByteBuffer buf) {
b = buf;
}
@Override
public int available() {
return b.remaining();
}
@Override
public int read() throws IOException {
return b.hasRemaining() ? (b.get() & 0xFF) : -1;
}
@Override
public int read(byte[] bytes, int off, int len) throws IOException {
if (!b.hasRemaining()) return -1;
len = Math.min(len, b.remaining());
b.get(bytes, off, len);
return len;
}
}

View File

@@ -1,20 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* Miscellaneous utility classes, such as {@code InputStream}/{@code OutputStream} manipulation utilities.
*/
package org.springframework.util;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.web.reactive.method.annotation;
import java.nio.ByteBuffer;
import java.util.List;
import org.reactivestreams.Publisher;
@@ -27,6 +26,7 @@ import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.RequestBody;
@@ -64,7 +64,7 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
mediaType = MediaType.APPLICATION_OCTET_STREAM;
}
ResolvableType type = ResolvableType.forMethodParameter(parameter);
Flux<ByteBuffer> body = exchange.getRequest().getBody();
Flux<DataBuffer> body = exchange.getRequest().getBody();
Flux<?> elementFlux = body;
ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Mono;
@@ -35,6 +36,8 @@ import org.springframework.core.codec.support.JsonObjectDecoder;
import org.springframework.core.codec.support.StringDecoder;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.NettyDataBufferAllocator;
import org.springframework.util.ObjectUtils;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.method.annotation.ExceptionHandlerMethodResolver;
@@ -57,6 +60,9 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
private ConversionService conversionService = new DefaultConversionService();
private DataBufferAllocator allocator =
new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false));
private final Map<Class<?>, ExceptionHandlerMethodResolver> exceptionHandlerCache =
new ConcurrentHashMap<Class<?>, ExceptionHandlerMethodResolver>(64);
@@ -85,13 +91,17 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
return this.conversionService;
}
public void setAllocator(DataBufferAllocator allocator) {
this.allocator = allocator;
}
@Override
public void afterPropertiesSet() throws Exception {
if (ObjectUtils.isEmpty(this.argumentResolvers)) {
List<Decoder<?>> decoders = Arrays.asList(new ByteBufferDecoder(),
new StringDecoder(), new JacksonJsonDecoder(new JsonObjectDecoder()));
new StringDecoder(allocator),
new JacksonJsonDecoder(new JsonObjectDecoder(allocator)));
this.argumentResolvers.add(new RequestParamArgumentResolver());
this.argumentResolvers.add(new RequestBodyArgumentResolver(decoders, this.conversionService));

View File

@@ -0,0 +1,59 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.codec.support;
import java.nio.charset.StandardCharsets;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.core.io.buffer.NettyDataBufferAllocator;
/**
* @author Arjen Poutsma
*/
@RunWith(Parameterized.class)
public abstract class AbstractAllocatingTestCase {
@Parameterized.Parameter
public DataBufferAllocator allocator;
@Parameterized.Parameters(name = "{0}")
public static Object[][] allocators() {
return new Object[][]{
{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(true))},
{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false))},
{new NettyDataBufferAllocator(new PooledByteBufAllocator(true))},
{new NettyDataBufferAllocator(new PooledByteBufAllocator(false))},
{new DefaultDataBufferAllocator(true)},
{new DefaultDataBufferAllocator(false)}
};
}
protected DataBuffer stringBuffer(String value) {
byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
DataBuffer buffer = allocator.allocateBuffer(bytes.length);
buffer.write(bytes);
return buffer;
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.decoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.List;
@@ -26,7 +26,7 @@ import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.ByteBufferDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import static java.util.stream.Collectors.toList;
@@ -35,7 +35,7 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class ByteBufferDecoderTests {
public class ByteBufferDecoderTests extends AbstractAllocatingTestCase {
private final ByteBufferDecoder decoder = new ByteBufferDecoder();
@@ -48,14 +48,25 @@ public class ByteBufferDecoderTests {
@Test
public void decode() throws InterruptedException {
ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer();
ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer();
Flux<ByteBuffer> source = Flux.just(fooBuffer, barBuffer);
DataBuffer fooBuffer = stringBuffer("foo");
DataBuffer barBuffer = stringBuffer("bar");
Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
Flux<ByteBuffer> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
List<ByteBuffer> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals(fooBuffer, results.get(0));
assertEquals(barBuffer, results.get(1));
assertBufferEquals(fooBuffer, results.get(0));
assertBufferEquals(barBuffer, results.get(1));
}
public void assertBufferEquals(DataBuffer expected, ByteBuffer actual) {
byte[] byteBufferBytes = new byte[actual.remaining()];
actual.get(byteBufferBytes);
byte[] dataBufferBytes = new byte[expected.readableByteCount()];
expected.read(dataBufferBytes);
assertArrayEquals(dataBufferBytes, byteBufferBytes);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,19 +14,20 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.encoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.ByteBufferEncoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import static java.util.stream.Collectors.toList;
@@ -35,27 +36,47 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class ByteBufferEncoderTests {
public class ByteBufferEncoderTests extends AbstractAllocatingTestCase {
private final ByteBufferEncoder encoder = new ByteBufferEncoder();
private ByteBufferEncoder encoder;
@Before
public void createEncoder() {
encoder = new ByteBufferEncoder(allocator);
}
@Test
public void canDecode() {
public void canEncode() {
assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.TEXT_PLAIN));
assertFalse(encoder.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.APPLICATION_JSON));
}
@Test
public void decode() throws InterruptedException {
ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer();
ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer();
Flux<ByteBuffer> source = Flux.just(fooBuffer, barBuffer);
Flux<ByteBuffer> output = encoder.encode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
List<ByteBuffer> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
public void encode() throws Exception {
byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
Flux<ByteBuffer> source =
Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes));
Flux<DataBuffer> output = encoder.encode(source,
ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
null);
List<DataBuffer> results =
StreamSupport.stream(output.toIterable().spliterator(), false)
.collect(toList());
assertEquals(2, results.size());
assertEquals(fooBuffer, results.get(0));
assertEquals(barBuffer, results.get(1));
assertEquals(3, results.get(0).readableByteCount());
assertEquals(3, results.get(1).readableByteCount());
byte[] buf = new byte[3];
results.get(0).read(buf);
assertArrayEquals(fooBytes, buf);
results.get(1).read(buf);
assertArrayEquals(barBytes, buf);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,9 +14,8 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.decoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
@@ -25,9 +24,8 @@ import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.JacksonJsonDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.reactive.codec.Pojo;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
@@ -35,7 +33,7 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class JacksonJsonDecoderTests {
public class JacksonJsonDecoderTests extends AbstractAllocatingTestCase {
private final JacksonJsonDecoder decoder = new JacksonJsonDecoder();
@@ -47,7 +45,8 @@ public class JacksonJsonDecoderTests {
@Test
public void decode() throws InterruptedException {
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Flux<DataBuffer> source =
Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,18 +14,17 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.encoder;
package org.springframework.core.codec.support;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.core.codec.support.JacksonJsonEncoder;
import org.springframework.http.MediaType;
import org.springframework.reactive.codec.Pojo;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
@@ -33,9 +32,14 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class JacksonJsonEncoderTests {
public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase {
private final JacksonJsonEncoder encoder = new JacksonJsonEncoder();
private JacksonJsonEncoder encoder;
@Before
public void createEncoder() {
encoder = new JacksonJsonEncoder(allocator);
}
@Test
public void canWrite() {
@@ -47,8 +51,8 @@ public class JacksonJsonEncoderTests {
public void write() throws InterruptedException {
Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,9 +14,8 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.decoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
@@ -25,9 +24,8 @@ import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.Jaxb2Decoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.reactive.codec.Pojo;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
@@ -35,7 +33,7 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class Jaxb2DecoderTests {
public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
private final Jaxb2Decoder decoder = new Jaxb2Decoder();
@@ -48,7 +46,8 @@ public class Jaxb2DecoderTests {
@Test
public void decode() throws InterruptedException {
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>").byteBuffer());
Flux<DataBuffer> source = Flux.just(stringBuffer(
"<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"));
Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,18 +14,17 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.encoder;
package org.springframework.core.codec.support;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.core.codec.support.Jaxb2Encoder;
import org.springframework.http.MediaType;
import org.springframework.reactive.codec.Pojo;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
@@ -33,9 +32,14 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class Jaxb2EncoderTests {
public class Jaxb2EncoderTests extends AbstractAllocatingTestCase {
private final Jaxb2Encoder encoder = new Jaxb2Encoder();
private Jaxb2Encoder encoder;
@Before
public void createEncoder() {
encoder = new Jaxb2Encoder(allocator);
}
@Test
public void canEncode() {
@@ -48,8 +52,8 @@ public class Jaxb2EncoderTests {
public void encode() throws InterruptedException {
Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,18 +14,16 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.decoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.io.buffer.Buffer;
import org.springframework.core.codec.support.JsonObjectDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
@@ -33,17 +31,16 @@ import static org.junit.Assert.assertEquals;
/**
* @author Sebastien Deleuze
*/
public class JsonObjectDecoderTests {
public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
@Test
public void decodeSingleChunkToJsonObject() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
Flux<DataBuffer> source =
Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
Flux<String> output =
decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
@@ -51,13 +48,11 @@ public class JsonObjectDecoderTests {
@Test
public void decodeMultipleChunksToJsonObject() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\"").byteBuffer(), Buffer.wrap(", \"bar\": \"barbar\"}").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\": \"foofoo\""),
stringBuffer(", \"bar\": \"barbar\"}"));
Flux<String> output =
decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
@@ -65,13 +60,12 @@ public class JsonObjectDecoderTests {
@Test
public void decodeSingleChunkToArray() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
Flux<DataBuffer> source = Flux.just(stringBuffer(
"[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
Flux<String> output =
decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
@@ -80,17 +74,22 @@ public class JsonObjectDecoderTests {
@Test
public void decodeMultipleChunksToArray() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
Flux<DataBuffer> source =
Flux.just(stringBuffer("[{\"foo\": \"foofoo\", \"bar\""), stringBuffer(
": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
Flux<String> output =
decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
}
private static String toString(DataBuffer buffer) {
byte[] b = new byte[buffer.readableByteCount()];
buffer.read(b);
return new String(b, StandardCharsets.UTF_8);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,45 +14,52 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.encoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import org.springframework.core.codec.support.JsonObjectEncoder;
import org.springframework.core.io.buffer.DataBuffer;
import static org.junit.Assert.assertEquals;
/**
* @author Sebastien Deleuze
*/
public class JsonObjectEncoderTests {
public class JsonObjectEncoderTests extends AbstractAllocatingTestCase {
private JsonObjectEncoder encoder;
@Before
public void createEncoder() {
encoder = new JsonObjectEncoder(allocator);
}
@Test
public void encodeSingleElementFlux() throws InterruptedException {
JsonObjectEncoder encoder = new JsonObjectEncoder();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Flux<DataBuffer> source =
Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
}).toIterable();
String result = String.join("", results);
assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"}]", result);
}
@Test
public void encodeSingleElementMono() throws InterruptedException {
JsonObjectEncoder encoder = new JsonObjectEncoder();
Mono<ByteBuffer> source = Mono.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Mono<DataBuffer> source =
Mono.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
}).toIterable();
String result = String.join("", results);
@@ -61,13 +68,12 @@ public class JsonObjectEncoderTests {
@Test
public void encodeTwoElementsFlux() throws InterruptedException {
JsonObjectEncoder encoder = new JsonObjectEncoder();
Flux<ByteBuffer> source = Flux.just(
Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(),
Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer());
Flux<DataBuffer> source =
Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"),
stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"));
Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
}).toIterable();
String result = String.join("", results);
@@ -76,15 +82,15 @@ public class JsonObjectEncoderTests {
@Test
public void encodeThreeElementsFlux() throws InterruptedException {
JsonObjectEncoder encoder = new JsonObjectEncoder();
Flux<ByteBuffer> source = Flux.just(
Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(),
Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer(),
Buffer.wrap("{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}").byteBuffer()
Flux<DataBuffer> source =
Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"),
stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"),
stringBuffer(
"{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}")
);
Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
}).toIterable();
String result = String.join("", results);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.reactive.codec;
package org.springframework.core.codec.support;
import javax.xml.bind.annotation.XmlRootElement;
@@ -52,4 +52,20 @@ public class Pojo {
this.bar = bar;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o instanceof Pojo) {
Pojo other = (Pojo) o;
return this.foo.equals(other.foo) && this.bar.equals(other.bar);
}
return false;
}
@Override
public int hashCode() {
return 31 * foo.hashCode() + bar.hashCode();
}
}

View File

@@ -14,12 +14,12 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.decoder;
package org.springframework.core.codec.support;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -28,7 +28,7 @@ import reactor.io.buffer.Buffer;
import rx.Single;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import static java.util.stream.Collectors.toList;
@@ -38,9 +38,15 @@ import static org.junit.Assert.*;
* @author Sebastien Deleuze
* @author Brian Clozel
*/
public class StringDecoderTests {
public class StringDecoderTests extends AbstractAllocatingTestCase {
private StringDecoder decoder;
@Before
public void createEncoder() {
decoder = new StringDecoder(allocator);
}
private final StringDecoder decoder = new StringDecoder();
@Test
public void canDecode() {
@@ -51,7 +57,7 @@ public class StringDecoderTests {
@Test
public void decode() throws InterruptedException {
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
Flux<String> output = this.decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
@@ -60,8 +66,8 @@ public class StringDecoderTests {
@Test
public void decodeDoNotBuffer() throws InterruptedException {
StringDecoder decoder = new StringDecoder(false);
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
StringDecoder decoder = new StringDecoder(allocator, false);
Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
Flux<String> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
@@ -71,7 +77,7 @@ public class StringDecoderTests {
@Test
public void decodeMono() throws InterruptedException {
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
Mono<String> mono = Mono.from(this.decoder.decode(source,
ResolvableType.forClassWithGenerics(Mono.class, String.class),
MediaType.TEXT_PLAIN));
@@ -81,7 +87,7 @@ public class StringDecoderTests {
@Test
public void decodeSingle() throws InterruptedException {
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"));
Single<String> single = RxJava1SingleConverter.from(this.decoder.decode(source,
ResolvableType.forClassWithGenerics(Single.class, String.class),
MediaType.TEXT_PLAIN));

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,17 +14,19 @@
* limitations under the License.
*/
package org.springframework.reactive.codec.encoder;
package org.springframework.core.codec.support;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.StringEncoder;
import org.springframework.http.MediaType;
import static java.util.stream.Collectors.toList;
@@ -33,9 +35,15 @@ import static org.junit.Assert.*;
/**
* @author Sebastien Deleuze
*/
public class StringEncoderTests {
@RunWith(Parameterized.class)
public class StringEncoderTests extends AbstractAllocatingTestCase {
private final StringEncoder encoder = new StringEncoder();
private StringEncoder encoder;
@Before
public void createEncoder() {
encoder = new StringEncoder(allocator);
}
@Test
public void canWrite() {
@@ -47,8 +55,8 @@ public class StringEncoderTests {
@Test
public void write() throws InterruptedException {
Flux<String> output = Flux.from(encoder.encode(Flux.just("foo"), null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
byte[] b = new byte[chunk.readableByteCount()];
chunk.read(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());

View File

@@ -0,0 +1,201 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.core.io.buffer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
/**
* @author Arjen Poutsma
*/
@RunWith(Parameterized.class)
public class DataBufferTests {
@Parameterized.Parameter
public DataBufferAllocator allocator;
@Parameterized.Parameters(name = "{0}")
public static Object[][] buffers() {
return new Object[][]{
{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(true))},
{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false))},
{new NettyDataBufferAllocator(new PooledByteBufAllocator(true))},
{new NettyDataBufferAllocator(new PooledByteBufAllocator(false))},
{new DefaultDataBufferAllocator(true)},
{new DefaultDataBufferAllocator(false)}};
}
private DataBuffer createDataBuffer(int capacity) {
return allocator.allocateBuffer(capacity);
}
@Test
public void writeAndRead() {
DataBuffer buffer = createDataBuffer(5);
buffer.write(new byte[]{'a', 'b', 'c'});
int ch = buffer.read();
assertEquals('a', ch);
buffer.write((byte) 'd');
buffer.write((byte) 'e');
byte[] result = new byte[4];
buffer.read(result);
assertArrayEquals(new byte[]{'b', 'c', 'd', 'e'}, result);
}
@Test
public void inputStream() throws IOException {
byte[] data = new byte[]{'a', 'b', 'c', 'd', 'e'};
DataBuffer buffer = createDataBuffer(4);
buffer.write(data);
buffer.read(); // readIndex++
InputStream inputStream = buffer.asInputStream();
int available = inputStream.available();
assertEquals(4, available);
int result = inputStream.read();
assertEquals('b', result);
available = inputStream.available();
assertEquals(3, available);
byte[] bytes = new byte[2];
int len = inputStream.read(bytes);
assertEquals(2, len);
assertArrayEquals(new byte[]{'c', 'd'}, bytes);
Arrays.fill(bytes, (byte) 0);
len = inputStream.read(bytes);
assertEquals(1, len);
assertArrayEquals(new byte[]{'e', (byte) 0}, bytes);
}
@Test
public void outputStream() throws IOException {
DataBuffer buffer = createDataBuffer(4);
buffer.write((byte) 'a');
OutputStream outputStream = buffer.asOutputStream();
outputStream.write(new byte[]{'b', 'c', 'd'});
buffer.write((byte) 'e');
byte[] bytes = new byte[5];
buffer.read(bytes);
assertArrayEquals(new byte[]{'a', 'b', 'c', 'd', 'e'}, bytes);
}
@Test
public void expand() {
DataBuffer buffer = createDataBuffer(1);
buffer.write((byte) 'a');
buffer.write((byte) 'b');
byte[] result = new byte[2];
buffer.read(result);
assertArrayEquals(new byte[]{'a', 'b'}, result);
buffer.write(new byte[]{'c', 'd'});
result = new byte[2];
buffer.read(result);
assertArrayEquals(new byte[]{'c', 'd'}, result);
}
@Test
public void writeByteBuffer() {
DataBuffer buffer1 = createDataBuffer(1);
buffer1.write((byte) 'a');
ByteBuffer buffer2 = createByteBuffer(2);
buffer2.put((byte) 'b');
buffer2.flip();
ByteBuffer buffer3 = createByteBuffer(3);
buffer3.put((byte) 'c');
buffer3.flip();
buffer1.write(buffer2, buffer3);
buffer1.write((byte) 'd'); // make sure the write index is correctly set
assertEquals(4, buffer1.readableByteCount());
byte[] result = new byte[4];
buffer1.read(result);
assertArrayEquals(new byte[]{'a', 'b', 'c', 'd'}, result);
}
@Test
public void writeDataBuffer() {
DataBuffer buffer1 = createDataBuffer(1);
buffer1.write((byte) 'a');
DataBuffer buffer2 = createDataBuffer(2);
buffer2.write((byte) 'b');
DataBuffer buffer3 = createDataBuffer(3);
buffer3.write((byte) 'c');
buffer1.write(buffer2, buffer3);
buffer1.write((byte) 'd'); // make sure the write index is correctly set
assertEquals(4, buffer1.readableByteCount());
byte[] result = new byte[4];
buffer1.read(result);
assertArrayEquals(new byte[]{'a', 'b', 'c', 'd'}, result);
}
private ByteBuffer createByteBuffer(int capacity) {
return ByteBuffer.allocate(capacity);
}
@Test
public void asByteBuffer() {
DataBuffer buffer = createDataBuffer(4);
buffer.write(new byte[]{'a', 'b', 'c'});
buffer.read(); // skip a
ByteBuffer result = buffer.asByteBuffer();
buffer.write((byte) 'd');
assertEquals(2, result.remaining());
byte[] resultBytes = new byte[2];
buffer.read(resultBytes);
assertArrayEquals(new byte[]{'b', 'c'}, resultBytes);
}
}

View File

@@ -1,11 +1,11 @@
/*
* Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -31,6 +31,8 @@ import reactor.core.timer.Timers;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
import org.springframework.http.server.reactive.boot.HttpServer;
@@ -52,6 +54,8 @@ public class AsyncIntegrationTests {
private final ProcessorGroup asyncGroup = Processors.asyncGroup();
private final DataBufferAllocator allocator = new DefaultDataBufferAllocator();
protected int port;
@Parameterized.Parameter(0)
@@ -109,7 +113,7 @@ public class AsyncIntegrationTests {
.dispatchOn(asyncGroup)
.collect(Buffer::new, Buffer::append)
.doOnSuccess(Buffer::flip)
.map(Buffer::byteBuffer)
.map((bytes) -> allocator.wrap(bytes.byteBuffer()))
);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@
package org.springframework.http.server.reactive;
import java.net.URI;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -35,7 +35,7 @@ public class MockServerHttpRequest implements ServerHttpRequest {
private HttpHeaders headers = new HttpHeaders();
private Flux<ByteBuffer> body;
private Flux<DataBuffer> body;
public MockServerHttpRequest(HttpMethod httpMethod, URI uri) {
@@ -43,7 +43,8 @@ public class MockServerHttpRequest implements ServerHttpRequest {
this.uri = uri;
}
public MockServerHttpRequest(Publisher<ByteBuffer> body, HttpMethod httpMethod, URI uri) {
public MockServerHttpRequest(Publisher<DataBuffer> body, HttpMethod httpMethod,
URI uri) {
this.body = Flux.from(body);
this.httpMethod = httpMethod;
this.uri = uri;
@@ -78,11 +79,11 @@ public class MockServerHttpRequest implements ServerHttpRequest {
}
@Override
public Flux<ByteBuffer> getBody() {
public Flux<DataBuffer> getBody() {
return this.body;
}
public void setBody(Publisher<ByteBuffer> body) {
public void setBody(Publisher<DataBuffer> body) {
this.body = Flux.from(body);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -15,12 +15,12 @@
*/
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
@@ -33,7 +33,7 @@ public class MockServerHttpResponse implements ServerHttpResponse {
private HttpHeaders headers = new HttpHeaders();
private Publisher<ByteBuffer> body;
private Publisher<DataBuffer> body;
@Override
@@ -51,12 +51,12 @@ public class MockServerHttpResponse implements ServerHttpResponse {
}
@Override
public Mono<Void> setBody(Publisher<ByteBuffer> body) {
public Mono<Void> setBody(Publisher<DataBuffer> body) {
this.body = body;
return Flux.from(body).after();
}
public Publisher<ByteBuffer> getBody() {
public Publisher<DataBuffer> getBody() {
return this.body;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
package org.springframework.http.server.reactive;
import java.nio.ByteBuffer;
import java.util.Random;
import org.apache.commons.logging.Log;
@@ -24,7 +23,9 @@ import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import static org.junit.Assert.assertEquals;
@@ -42,7 +43,7 @@ public class RandomHandler implements HttpHandler {
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
request.getBody().subscribe(new Subscriber<ByteBuffer>() {
request.getBody().subscribe(new Subscriber<DataBuffer>() {
private Subscription s;
private int requestSize = 0;
@@ -54,8 +55,8 @@ public class RandomHandler implements HttpHandler {
}
@Override
public void onNext(ByteBuffer bytes) {
requestSize += new Buffer(bytes).limit();
public void onNext(DataBuffer bytes) {
requestSize += bytes.readableByteCount();
}
@Override
@@ -72,7 +73,11 @@ public class RandomHandler implements HttpHandler {
});
response.getHeaders().setContentLength(RESPONSE_SIZE);
return response.setBody(Mono.just(ByteBuffer.wrap(randomBytes())));
byte[] randomBytes = randomBytes();
DataBuffer buffer =
new DefaultDataBufferAllocator().allocateBuffer(randomBytes.length);
buffer.write(randomBytes);
return response.setBody(Mono.just(buffer));
}
private byte[] randomBytes() {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
package org.springframework.http.server.reactive;
import java.io.InputStream;
import java.io.OutputStream;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
@@ -26,9 +28,10 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.http.MediaType;
import org.springframework.util.BufferOutputStream;
import org.springframework.util.ByteBufferPublisherInputStream;
import static org.junit.Assert.fail;
@@ -49,7 +52,7 @@ public class XmlHandler implements HttpHandler {
Runnable r = () -> {
try {
ByteBufferPublisherInputStream bis = new ByteBufferPublisherInputStream(request.getBody());
InputStream bis = DataBufferUtils.toInputStream(request.getBody());
XmlHandlerIntegrationTests.Person johnDoe =
(XmlHandlerIntegrationTests.Person) unmarshaller.unmarshal(bis);
@@ -67,13 +70,13 @@ public class XmlHandler implements HttpHandler {
response.getHeaders().setContentType(MediaType.APPLICATION_XML);
XmlHandlerIntegrationTests.Person janeDoe = new XmlHandlerIntegrationTests.Person("Jane Doe");
Buffer buffer = new Buffer();
BufferOutputStream bos = new BufferOutputStream(buffer);
DataBuffer buffer = new DefaultDataBufferAllocator().allocateBuffer();
OutputStream bos = buffer.asOutputStream();
marshaller.marshal(janeDoe, bos);
bos.close();
buffer.flip();
return response.setBody(Flux.just(buffer.byteBuffer()));
return response.setBody(Flux.just(buffer));
}
catch (Exception ex) {
logger.error(ex, ex);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -13,10 +13,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.web.reactive;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
@@ -33,6 +33,8 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.core.codec.Encoder;
import org.springframework.core.codec.support.StringEncoder;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
@@ -109,7 +111,7 @@ public class DispatcherHandlerErrorTests {
@Test
public void noResolverForArgument() throws Exception {
this.request.setUri(new URI("/uknown-argument-type"));
this.request.setUri(new URI("/unknown-argument-type"));
Publisher<Void> publisher = this.dispatcherHandler.handle(this.exchange);
Throwable ex = awaitErrorSignal(publisher);
@@ -153,7 +155,9 @@ public class DispatcherHandlerErrorTests {
public void notAcceptable() throws Exception {
this.request.setUri(new URI("/request-body"));
this.request.getHeaders().setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
this.request.setBody(Mono.just(ByteBuffer.wrap("body".getBytes("UTF-8"))));
DataBuffer buffer = new DefaultDataBufferAllocator().allocateBuffer()
.write("body".getBytes("UTF-8"));
this.request.setBody(Mono.just(buffer));
Publisher<Void> publisher = this.dispatcherHandler.handle(this.exchange);
Throwable ex = awaitErrorSignal(publisher);
@@ -178,7 +182,7 @@ public class DispatcherHandlerErrorTests {
@Test
public void dispatcherHandlerWithHttpExceptionHandler() throws Exception {
this.request.setUri(new URI("/uknown-argument-type"));
this.request.setUri(new URI("/unknown-argument-type"));
WebExceptionHandler exceptionHandler = new ServerError500ExceptionHandler();
WebHandler webHandler = new ExceptionHandlingWebHandler(this.dispatcherHandler, exceptionHandler);
@@ -190,7 +194,7 @@ public class DispatcherHandlerErrorTests {
@Test
public void filterChainWithHttpExceptionHandler() throws Exception {
this.request.setUri(new URI("/uknown-argument-type"));
this.request.setUri(new URI("/unknown-argument-type"));
WebHandler webHandler = new FilteringWebHandler(this.dispatcherHandler, new TestWebFilter());
webHandler = new ExceptionHandlingWebHandler(webHandler, new ServerError500ExceptionHandler());
@@ -224,7 +228,8 @@ public class DispatcherHandlerErrorTests {
@Bean
public ResponseBodyResultHandler resultHandler() {
List<Encoder<?>> encoders = Collections.singletonList(new StringEncoder());
List<Encoder<?>> encoders = Collections
.singletonList(new StringEncoder(new DefaultDataBufferAllocator()));
return new ResponseBodyResultHandler(encoders, new DefaultConversionService());
}
@@ -238,8 +243,8 @@ public class DispatcherHandlerErrorTests {
@SuppressWarnings("unused")
private static class TestController {
@RequestMapping("/uknown-argument-type")
public void uknownArgumentType(Foo arg) {
@RequestMapping("/unknown-argument-type")
public void unknownArgumentType(Foo arg) {
}
@RequestMapping("/error-signal")

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.web.reactive.handler;
import java.net.URI;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@@ -27,6 +28,8 @@ import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.HttpStatus;
import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
@@ -139,7 +142,9 @@ public class SimpleUrlHandlerMappingIntegrationTests extends AbstractHttpHandler
@Override
public Mono<Void> handle(WebServerExchange exchange) {
return exchange.getResponse().setBody(Flux.just(Buffer.wrap("foo").byteBuffer()));
DataBuffer buffer = new DefaultDataBufferAllocator().allocateBuffer()
.write("foo".getBytes(StandardCharsets.UTF_8));
return exchange.getResponse().setBody(Flux.just(buffer));
}
}
@@ -147,7 +152,9 @@ public class SimpleUrlHandlerMappingIntegrationTests extends AbstractHttpHandler
@Override
public Mono<Void> handle(WebServerExchange exchange) {
return exchange.getResponse().setBody(Flux.just(Buffer.wrap("bar").byteBuffer()));
DataBuffer buffer = new DefaultDataBufferAllocator().allocateBuffer()
.write("bar".getBytes(StandardCharsets.UTF_8));
return exchange.getResponse().setBody(Flux.just(buffer));
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -48,6 +48,9 @@ import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.core.convert.support.ReactiveStreamsToCompletableFutureConverter;
import org.springframework.core.convert.support.ReactiveStreamsToReactorStreamConverter;
import org.springframework.core.convert.support.ReactiveStreamsToRxJava1Converter;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.RequestEntity;
@@ -380,8 +383,10 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
@Bean
public ResponseBodyResultHandler responseBodyResultHandler() {
DataBufferAllocator allocator = new DefaultDataBufferAllocator();
return new ResponseBodyResultHandler(Arrays.asList(
new ByteBufferEncoder(), new StringEncoder(), new JacksonJsonEncoder(new JsonObjectEncoder())),
new ByteBufferEncoder(allocator), new StringEncoder(allocator),
new JacksonJsonEncoder(allocator, new JsonObjectEncoder(allocator))),
conversionService());
}
@@ -426,9 +431,9 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
@RequestMapping("/raw")
public Publisher<ByteBuffer> rawResponseBody() {
JacksonJsonEncoder encoder = new JacksonJsonEncoder();
JacksonJsonEncoder encoder = new JacksonJsonEncoder(new DefaultDataBufferAllocator());
return encoder.encode(Stream.just(new Person("Robert")),
ResolvableType.forClass(Person.class), MediaType.APPLICATION_JSON);
ResolvableType.forClass(Person.class), MediaType.APPLICATION_JSON).map(DataBuffer::asByteBuffer);
}
@RequestMapping("/stream-result")

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,11 +22,12 @@ import org.junit.Test;
import org.reactivestreams.Publisher;
import org.springframework.core.ResolvableType;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.codec.support.StringEncoder;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.reactive.HandlerResult;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -40,7 +41,8 @@ public class ResponseBodyResultHandlerTests {
@Test
public void supports() throws NoSuchMethodException {
ResponseBodyResultHandler handler = new ResponseBodyResultHandler(Collections.singletonList(
new StringEncoder()), new DefaultConversionService());
new StringEncoder(new DefaultDataBufferAllocator())),
new DefaultConversionService());
TestController controller = new TestController();
HandlerMethod hm = new HandlerMethod(controller,TestController.class.getMethod("notAnnotated"));