Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Minor revision of reactive support layout (ahead of 5.0 M1)

Browse Source 
DataSourceUtils moved to main core.io.buffer package.
Consistently named Jackson2JsonDecoder/Encoder and Jaxb2XmlDecoder/Encoder.
Plenty of related polishing.
This commit is contained in:
Juergen Hoeller
2016-07-26 15:39:32 +02:00
parent 3d6a5bcd66
commit c13f8419f9
51 changed files with 335 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-core/src/main/java/org/springframework/core/codec/AbstractDecoder.java
View File

@@ -17,7 +17,6 @@
package org.springframework.core.codec;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.reactivestreams.Publisher;
@@ -36,7 +35,7 @@ import org.springframework.util.MimeType;
*/
public abstract class AbstractDecoder<T> implements Decoder<T> {
private List<MimeType> decodableMimeTypes = Collections.emptyList();
private final List<MimeType> decodableMimeTypes;
protected AbstractDecoder(MimeType... supportedMimeTypes) {
@@ -54,7 +53,7 @@ public abstract class AbstractDecoder<T> implements Decoder<T> {
if (mimeType == null) {
return true;
}
return this.decodableMimeTypes.stream().anyMatch(m -> m.isCompatibleWith(mimeType));
return this.decodableMimeTypes.stream().anyMatch(candidate -> candidate.isCompatibleWith(mimeType));
}
@Override

5
spring-core/src/main/java/org/springframework/core/codec/AbstractEncoder.java
View File

@@ -17,7 +17,6 @@
package org.springframework.core.codec;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.springframework.core.ResolvableType;
@@ -32,7 +31,7 @@ import org.springframework.util.MimeType;
*/
public abstract class AbstractEncoder<T> implements Encoder<T> {
private List<MimeType> encodableMimeTypes = Collections.emptyList();
private final List<MimeType> encodableMimeTypes;
protected AbstractEncoder(MimeType... supportedMimeTypes) {
@@ -50,7 +49,7 @@ public abstract class AbstractEncoder<T> implements Encoder<T> {
if (mimeType == null) {
return true;
}
return this.encodableMimeTypes.stream().anyMatch(m -> m.isCompatibleWith(mimeType));
return this.encodableMimeTypes.stream().anyMatch(candidate -> candidate.isCompatibleWith(mimeType));
}
}

5
spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java
View File

@@ -40,9 +40,8 @@ public abstract class AbstractSingleValueEncoder<T> extends AbstractEncoder<T> {
@Override
public final Flux<DataBuffer> encode(Publisher<? extends T> inputStream,
DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
Object... hints) {
public final Flux<DataBuffer> encode(Publisher<? extends T> inputStream, DataBufferFactory bufferFactory,
ResolvableType elementType, MimeType mimeType, Object... hints) {
return Flux.from(inputStream).
take(1).

5
spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java
View File

@@ -23,7 +23,7 @@ import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -36,7 +36,6 @@ import org.springframework.util.MimeTypeUtils;
*/
public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
public ByteBufferDecoder() {
super(MimeTypeUtils.ALL);
}
@@ -61,4 +60,4 @@ public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
});
}
}
}

3
spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java
View File

@@ -35,7 +35,6 @@ import org.springframework.util.MimeTypeUtils;
*/
public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
public ByteBufferEncoder() {
super(MimeTypeUtils.ALL);
}
@@ -55,4 +54,4 @@ public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
return Flux.from(inputStream).map(bufferFactory::wrap);
}
}
}

2
spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java
View File

@@ -30,7 +30,7 @@ import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.util.MimeType;
/**
* Encode from a CharSequence stream to a bytes stream.
* Encode from a {@code CharSequence} stream to a bytes stream.
*
* @author Sebastien Deleuze
* @author Arjen Poutsma

10
spring-core/src/main/java/org/springframework/core/codec/CodecException.java
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,12 +27,12 @@ import org.springframework.core.NestedRuntimeException;
@SuppressWarnings("serial")
public class CodecException extends NestedRuntimeException {
public CodecException(String msg, Throwable cause) {
super(msg, cause);
}
public CodecException(String msg) {
super(msg);
}
public CodecException(String msg, Throwable cause) {
super(msg, cause);
}
}

5
spring-core/src/main/java/org/springframework/core/codec/Decoder.java
View File

@@ -32,15 +32,14 @@ import org.springframework.util.MimeType;
*
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @param <T> the type of elements in the output stream
* @since 5.0
* @param <T> the type of elements in the output stream
*/
public interface Decoder<T> {
/**
* Whether the decoder supports the given target element type and the MIME
* type of the source stream.
*
* @param elementType the target element type for the output stream
* @param mimeType the mime type associated with the stream to decode
* @param hints additional information about how to do decode, optional
@@ -50,7 +49,6 @@ public interface Decoder<T> {
/**
* Decode a {@link DataBuffer} input stream into a Flux of {@code T}.
*
* @param inputStream the {@code DataBuffer} input stream to decode
* @param elementType the expected type of elements in the output stream;
* this type must have been previously passed to the {@link #canDecode}
@@ -64,7 +62,6 @@ public interface Decoder<T> {
/**
* Decode a {@link DataBuffer} input stream into a Mono of {@code T}.
*
* @param inputStream the {@code DataBuffer} input stream to decode
* @param elementType the expected type of elements in the output stream;
* this type must have been previously passed to the {@link #canDecode}

4
spring-core/src/main/java/org/springframework/core/codec/Encoder.java
View File

@@ -32,15 +32,14 @@ import org.springframework.util.MimeType;
*
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @param <T> the type of elements in the input stream
* @since 5.0
* @param <T> the type of elements in the input stream
*/
public interface Encoder<T> {
/**
* Whether the encoder supports the given source element type and the MIME
* type for the output stream.
*
* @param elementType the type of elements in the source stream
* @param mimeType the MIME type for the output stream
* @param hints additional information about how to do encode, optional
@@ -51,7 +50,6 @@ public interface Encoder<T> {
/**
* Encode a stream of Objects of type {@code T} into a {@link DataBuffer}
* output stream.
*
* @param inputStream the input stream of Objects to encode
* @param bufferFactory for creating output stream {@code DataBuffer}'s
* @param elementType the expected type of elements in the input stream;

4
spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java
View File

@@ -27,7 +27,7 @@ import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.InputStreamResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -39,7 +39,6 @@ import org.springframework.util.MimeTypeUtils;
*/
public class ResourceDecoder extends AbstractDecoder<Resource> {
public ResourceDecoder() {
super(MimeTypeUtils.ALL);
}
@@ -79,4 +78,5 @@ public class ResourceDecoder extends AbstractDecoder<Resource> {
return Flux.error(new IllegalStateException("Unsupported resource class: " + clazz));
}
}
}

3
spring-core/src/main/java/org/springframework/core/codec/ResourceEncoder.java
View File

@@ -25,7 +25,7 @@ import org.springframework.core.ResolvableType;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
@@ -41,7 +41,6 @@ public class ResourceEncoder extends AbstractSingleValueEncoder<Resource> {
public static final int DEFAULT_BUFFER_SIZE = StreamUtils.BUFFER_SIZE;
private final int bufferSize;

16
spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java
View File

@@ -29,16 +29,16 @@ import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.support.DataBufferUtils;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
/**
* Decode from a bytes stream to a String stream.
* Decode from a bytes stream to a {@code String} stream.
*
* <p>By default, this decoder will split the received {@link DataBuffer}s along newline
* characters ({@code \r\n}), but this can be changed by passing {@code false} as
* constructor argument.
* <p>By default, this decoder will split the received {@link DataBuffer}s
* along newline characters ({@code \r\n}), but this can be changed by
* passing {@code false} as a constructor argument.
*
* @author Sebastien Deleuze
* @author Brian Clozel
@@ -59,7 +59,6 @@ public class StringDecoder extends AbstractDecoder<String> {
/**
* Create a {@code StringDecoder} that decodes a bytes stream to a String stream
*
* <p>By default, this decoder will split along new lines.
*/
public StringDecoder() {
@@ -68,7 +67,6 @@ public class StringDecoder extends AbstractDecoder<String> {
/**
* Create a {@code StringDecoder} that decodes a bytes stream to a String stream
*
* @param splitOnNewline whether this decoder should split the received data buffers
* along newline characters
*/
@@ -80,8 +78,8 @@ public class StringDecoder extends AbstractDecoder<String> {
@Override
public boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints) {
return super.canDecode(elementType, mimeType, hints) &&
String.class.equals(elementType.getRawClass());
return (super.canDecode(elementType, mimeType, hints) &&
String.class.equals(elementType.getRawClass()));
}
@Override

37
spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java
View File

@@ -30,13 +30,13 @@ import java.util.function.IntPredicate;
public interface DataBuffer {
/**
* Returns the {@link DataBufferFactory} that created this buffer.
* Return the {@link DataBufferFactory} that created this buffer.
* @return the creating buffer factory
*/
DataBufferFactory factory();
/**
* Returns the index of the first byte in this buffer that matches the given
* Return the index of the first byte in this buffer that matches the given
* predicate.
* @param predicate the predicate to match
* @param fromIndex the index to start the search from
@@ -46,7 +46,7 @@ public interface DataBuffer {
int indexOf(IntPredicate predicate, int fromIndex);
/**
* Returns the index of the last byte in this buffer that matches the given
* Return the index of the last byte in this buffer that matches the given
* predicate.
* @param predicate the predicate to match
* @param fromIndex the index to start the search from
@@ -56,28 +56,27 @@ public interface DataBuffer {
int lastIndexOf(IntPredicate predicate, int fromIndex);
/**
* Returns the number of bytes that can be read from this data buffer.
* Return 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.
* Read 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
* Read 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,
* Read 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
@@ -94,7 +93,7 @@ public interface DataBuffer {
DataBuffer write(byte b);
/**
* Writes the given source into this buffer, startin at the current writing position
* Write 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
@@ -102,7 +101,7 @@ public interface DataBuffer {
DataBuffer write(byte[] source);
/**
* Writes at most {@code length} bytes of the given source into this buffer, starting
* Write 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
@@ -112,7 +111,7 @@ public interface DataBuffer {
DataBuffer write(byte[] source, int offset, int length);
/**
* Writes one or more {@code DataBuffer}s to this buffer, starting at the current
* Write one or more {@code DataBuffer}s to this buffer, starting at the current
* writing position.
* @param buffers the byte buffers to write into this buffer
* @return this buffer
@@ -120,7 +119,7 @@ public interface DataBuffer {
DataBuffer write(DataBuffer... buffers);
/**
* Writes one or more {@link ByteBuffer} to this buffer, starting at the current
* Write 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
@@ -128,7 +127,7 @@ public interface DataBuffer {
DataBuffer write(ByteBuffer... buffers);
/**
* Creates a new {@code DataBuffer} whose contents is a shared subsequence of this
* Create a new {@code DataBuffer} whose contents is a shared subsequence of this
* data buffer's content. Data between this data buffer and the returned buffer is
* shared; though changes in the returned buffer's position will not be reflected
* in the reading nor writing position of this data buffer.
@@ -139,23 +138,23 @@ public interface DataBuffer {
DataBuffer slice(int index, int length);
/**
* 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 reading nor writing position of this data buffer.
* Expose 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 reading nor writing position 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
* Expose 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
* Expose 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
*/

2
spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferFactory.java
View File

@@ -23,8 +23,8 @@ import java.nio.ByteBuffer;
* data buffers.
*
* @author Arjen Poutsma
* @see DataBuffer
* @since 5.0
* @see DataBuffer
*/
public interface DataBufferFactory {

48
spring-core/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java → spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.core.io.buffer.support;
package org.springframework.core.io.buffer;
import java.io.IOException;
import java.io.InputStream;
@@ -29,9 +29,6 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.PooledDataBuffer;
import org.springframework.util.Assert;
/**i
@@ -48,12 +45,13 @@ public abstract class DataBufferUtils {
channel.close();
}
}
catch (IOException ignored) {
catch (IOException ex) {
}
};
/**
* Reads the given {@code InputStream} into a {@code Flux} of
* Read the given {@code InputStream} into a {@code Flux} of
* {@code DataBuffer}s. Closes the input stream when the flux is terminated.
* @param inputStream the input stream to read from
* @param dataBufferFactory the factory to create data buffers with
@@ -62,15 +60,16 @@ public abstract class DataBufferUtils {
*/
public static Flux<DataBuffer> read(InputStream inputStream,
DataBufferFactory dataBufferFactory, int bufferSize) {
Assert.notNull(inputStream, "'inputStream' must not be null");
Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
Assert.notNull(inputStream, "InputStream must not be null");
Assert.notNull(dataBufferFactory, "DataBufferFactory must not be null");
ReadableByteChannel channel = Channels.newChannel(inputStream);
return read(channel, dataBufferFactory, bufferSize);
}
/**
* Reads the given {@code ReadableByteChannel} into a {@code Flux} of
* Read the given {@code ReadableByteChannel} into a {@code Flux} of
* {@code DataBuffer}s. Closes the channel when the flux is terminated.
* @param channel the channel to read from
* @param dataBufferFactory the factory to create data buffers with
@@ -79,8 +78,9 @@ public abstract class DataBufferUtils {
*/
public static Flux<DataBuffer> read(ReadableByteChannel channel,
DataBufferFactory dataBufferFactory, int bufferSize) {
Assert.notNull(channel, "'channel' must not be null");
Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
Assert.notNull(channel, "ReadableByteChannel must not be null");
Assert.notNull(dataBufferFactory, "DataBufferFactory must not be null");
return Flux.generate(() -> channel,
new ReadableByteChannelGenerator(dataBufferFactory, bufferSize),
@@ -88,18 +88,16 @@ public abstract class DataBufferUtils {
}
/**
* Relays buffers from the given {@link Publisher} until the total
* {@linkplain DataBuffer#readableByteCount() byte count} reaches the given
* maximum byte count, or until the publisher is complete.
* Relay buffers from the given {@link Publisher} until the total
* {@linkplain DataBuffer#readableByteCount() byte count} reaches
* the given maximum byte count, or until the publisher is complete.
* @param publisher the publisher to filter
* @param maxByteCount the maximum byte count
* @return a flux whose maximum byte count is {@code maxByteCount}
*/
public static Flux<DataBuffer> takeUntilByteCount(Publisher<DataBuffer> publisher,
long maxByteCount) {
Assert.notNull(publisher, "'publisher' must not be null");
public static Flux<DataBuffer> takeUntilByteCount(Publisher<DataBuffer> publisher, long maxByteCount) {
Assert.notNull(publisher, "Publisher must not be null");
Assert.isTrue(maxByteCount >= 0, "'maxByteCount' must be a positive number");
AtomicLong byteCountDown = new AtomicLong(maxByteCount);
return Flux.from(publisher).
@@ -122,7 +120,7 @@ public abstract class DataBufferUtils {
}
/**
* Retains the given data buffer, it it is a {@link PooledDataBuffer}.
* Retain the given data buffer, it it is a {@link PooledDataBuffer}.
* @param dataBuffer the data buffer to retain
* @return the retained buffer
*/
@@ -137,7 +135,7 @@ public abstract class DataBufferUtils {
}
/**
* Releases the given data buffer, if it is a {@link PooledDataBuffer}.
* Release the given data buffer, if it is a {@link PooledDataBuffer}.
* @param dataBuffer the data buffer to release
* @return {@code true} if the buffer was released; {@code false} otherwise.
*/
@@ -148,23 +146,21 @@ public abstract class DataBufferUtils {
return false;
}
private static class ReadableByteChannelGenerator
implements BiFunction<ReadableByteChannel, SynchronousSink<DataBuffer>,
ReadableByteChannel> {
implements BiFunction<ReadableByteChannel, SynchronousSink<DataBuffer>, ReadableByteChannel> {
private final DataBufferFactory dataBufferFactory;
private final int chunkSize;
public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory,
int chunkSize) {
public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory, int chunkSize) {
this.dataBufferFactory = dataBufferFactory;
this.chunkSize = chunkSize;
}
@Override
public ReadableByteChannel apply(ReadableByteChannel
channel, SynchronousSink<DataBuffer> sub) {
public ReadableByteChannel apply(ReadableByteChannel channel, SynchronousSink<DataBuffer> sub) {
try {
ByteBuffer byteBuffer = ByteBuffer.allocate(chunkSize);
int read;

51
spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java
View File

@@ -46,6 +46,7 @@ public class DefaultDataBuffer implements DataBuffer {
private int writePosition;
/**
* Create a new {@code DefaultDataBuffer} based on the given
* {@code ByteBuffer}. Both reading and writing position of this buffer are
@@ -59,6 +60,7 @@ public class DefaultDataBuffer implements DataBuffer {
DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition,
DefaultDataBufferFactory dataBufferFactory) {
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");
@@ -72,10 +74,6 @@ public class DefaultDataBuffer implements DataBuffer {
this.dataBufferFactory = dataBufferFactory;
}
@Override
public DefaultDataBufferFactory factory() {
return this.dataBufferFactory;
}
/**
* Directly exposes the native {@code ByteBuffer} that this buffer is based on.
@@ -85,6 +83,11 @@ public class DefaultDataBuffer implements DataBuffer {
return this.byteBuffer;
}
@Override
public DefaultDataBufferFactory factory() {
return this.dataBufferFactory;
}
@Override
public int indexOf(IntPredicate predicate, int fromIndex) {
Assert.notNull(predicate, "'predicate' must not be null");
@@ -130,18 +133,14 @@ public class DefaultDataBuffer implements DataBuffer {
@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;
}
@@ -163,14 +162,12 @@ public class DefaultDataBuffer implements DataBuffer {
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;
@@ -179,7 +176,6 @@ public class DefaultDataBuffer implements DataBuffer {
@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;
@@ -199,15 +195,10 @@ public class DefaultDataBuffer implements DataBuffer {
@Override
public DefaultDataBuffer write(ByteBuffer... byteBuffers) {
Assert.notEmpty(byteBuffers, "'byteBuffers' must not be empty");
int extraCapacity =
Arrays.stream(byteBuffers).mapToInt(ByteBuffer::remaining).sum();
int extraCapacity = Arrays.stream(byteBuffers).mapToInt(ByteBuffer::remaining).sum();
ensureExtraCapacity(extraCapacity);
Arrays.stream(byteBuffers)
.forEach(byteBuffer -> writeInternal(buffer -> buffer.put(byteBuffer)));
return this;
}
@@ -279,23 +270,23 @@ public class DefaultDataBuffer implements DataBuffer {
}
@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.readPosition == other.readPosition &&
this.writePosition == other.writePosition &&
this.byteBuffer.equals(other.byteBuffer);
if (!(obj instanceof DefaultDataBuffer)) {
return false;
}
return false;
DefaultDataBuffer other = (DefaultDataBuffer) obj;
return (this.readPosition == other.readPosition &&
this.writePosition == other.writePosition &&
this.byteBuffer.equals(other.byteBuffer));
}
@Override
public int hashCode() {
return this.byteBuffer.hashCode();
}
@Override
@@ -303,6 +294,7 @@ public class DefaultDataBuffer implements DataBuffer {
return this.byteBuffer.toString();
}
private class DefaultDataBufferInputStream extends InputStream {
@Override
@@ -332,6 +324,7 @@ public class DefaultDataBuffer implements DataBuffer {
}
}
private class DefaultDataBufferOutputStream extends OutputStream {
@Override
@@ -347,6 +340,7 @@ public class DefaultDataBuffer implements DataBuffer {
}
}
private static class SlicedDefaultDataBuffer extends DefaultDataBuffer {
SlicedDefaultDataBuffer(ByteBuffer byteBuffer, int readPosition,
@@ -360,4 +354,5 @@ public class DefaultDataBuffer implements DataBuffer {
"Growing the capacity of a sliced buffer is not supported");
}
}
}

9
spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java
View File

@@ -70,8 +70,7 @@ public class DefaultDataBufferFactory implements DataBufferFactory {
* {@code false} otherwise
*/
public DefaultDataBufferFactory(boolean preferDirect, int defaultInitialCapacity) {
Assert.isTrue(defaultInitialCapacity > 0,
"'defaultInitialCapacity' should be larger than 0");
Assert.isTrue(defaultInitialCapacity > 0, "'defaultInitialCapacity' should be larger than 0");
this.preferDirect = preferDirect;
this.defaultInitialCapacity = defaultInitialCapacity;
}
@@ -84,9 +83,9 @@ public class DefaultDataBufferFactory implements DataBufferFactory {
@Override
public DefaultDataBuffer allocateBuffer(int initialCapacity) {
return this.preferDirect ?
return (this.preferDirect ?
new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity), this) :
new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity), this);
new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity), this));
}
@Override
@@ -97,7 +96,7 @@ public class DefaultDataBufferFactory implements DataBufferFactory {
@Override
public String toString() {
return "DefaultDataBufferFactory - preferDirect: " + this.preferDirect;
return "DefaultDataBufferFactory (preferDirect=" + this.preferDirect + ")";
}
}

39
spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
View File

@@ -44,6 +44,7 @@ public class NettyDataBuffer implements PooledDataBuffer {
private ByteBuf byteBuf;
/**
* Creates a new {@code NettyDataBuffer} based on the given {@code ByteBuff}.
* @param byteBuf the buffer to base this buffer on
@@ -56,10 +57,6 @@ public class NettyDataBuffer implements PooledDataBuffer {
this.dataBufferFactory = dataBufferFactory;
}
@Override
public NettyDataBufferFactory factory() {
return this.dataBufferFactory;
}
/**
* Directly exposes the native {@code ByteBuf} that this buffer is based on.
@@ -69,6 +66,11 @@ public class NettyDataBuffer implements PooledDataBuffer {
return this.byteBuf;
}
@Override
public NettyDataBufferFactory factory() {
return this.dataBufferFactory;
}
@Override
public int indexOf(IntPredicate predicate, int fromIndex) {
Assert.notNull(predicate, "'predicate' must not be null");
@@ -79,7 +81,6 @@ public class NettyDataBuffer implements PooledDataBuffer {
return -1;
}
int length = this.byteBuf.writerIndex() - fromIndex;
return this.byteBuf.forEachByte(fromIndex, length, predicate.negate()::test);
}
@@ -90,7 +91,6 @@ public class NettyDataBuffer implements PooledDataBuffer {
return -1;
}
fromIndex = Math.min(fromIndex, this.byteBuf.writerIndex() - 1);
return this.byteBuf.forEachByteDesc(0, fromIndex, predicate.negate()::test);
}
@@ -141,7 +141,6 @@ public class NettyDataBuffer implements PooledDataBuffer {
ByteBuf[] nativeBuffers = Arrays.stream(buffers)
.map(b -> ((NettyDataBuffer) b).getNativeBuffer())
.toArray(ByteBuf[]::new);
write(nativeBuffers);
}
else {
@@ -157,10 +156,8 @@ public class NettyDataBuffer implements PooledDataBuffer {
@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);
}
@@ -173,9 +170,8 @@ public class NettyDataBuffer implements PooledDataBuffer {
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);
CompositeByteBuf composite = new CompositeByteBuf(
this.byteBuf.alloc(), this.byteBuf.isDirect(), byteBufs.length + 1);
composite.addComponent(this.byteBuf);
composite.addComponents(byteBufs);
@@ -184,7 +180,6 @@ public class NettyDataBuffer implements PooledDataBuffer {
composite.writerIndex(writerIndex);
this.byteBuf = composite;
return this;
}
@@ -219,25 +214,27 @@ public class NettyDataBuffer implements PooledDataBuffer {
return this.byteBuf.release();
}
@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);
if (!(obj instanceof NettyDataBuffer)) {
return false;
}
return false;
NettyDataBuffer other = (NettyDataBuffer) obj;
return this.byteBuf.equals(other.byteBuf);
}
@Override
public int hashCode() {
return this.byteBuf.hashCode();
}
@Override
public String toString() {
return this.byteBuf.toString();
}
}

9
spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java
View File

@@ -25,8 +25,8 @@ import io.netty.buffer.Unpooled;
import org.springframework.util.Assert;
/**
* Implementation of the {@code DataBufferFactory} interface based on a Netty
* {@link ByteBufAllocator}.
* Implementation of the {@code DataBufferFactory} interface based on a
* Netty {@link ByteBufAllocator}.
*
* @author Arjen Poutsma
* @since 5.0
@@ -46,7 +46,6 @@ public class NettyDataBufferFactory implements DataBufferFactory {
*/
public NettyDataBufferFactory(ByteBufAllocator byteBufAllocator) {
Assert.notNull(byteBufAllocator, "'byteBufAllocator' must not be null");
this.byteBufAllocator = byteBufAllocator;
}
@@ -70,7 +69,7 @@ public class NettyDataBufferFactory implements DataBufferFactory {
}
/**
* Wraps the given Netty {@link ByteBuf} in a {@code NettyDataBuffer}.
* Wrap the given Netty {@link ByteBuf} in a {@code NettyDataBuffer}.
* @param byteBuf the Netty byte buffer to wrap
* @return the wrapped buffer
*/
@@ -79,7 +78,7 @@ public class NettyDataBufferFactory implements DataBufferFactory {
}
/**
* Returns the given Netty {@link DataBuffer} as a {@link ByteBuf}. Returns the
* Return the given Netty {@link DataBuffer} as a {@link ByteBuf}. Returns the
* {@linkplain NettyDataBuffer#getNativeBuffer() native buffer} if {@code buffer} is
* a {@link NettyDataBuffer}; returns {@link Unpooled#wrappedBuffer(ByteBuffer)}
* otherwise.

4
spring-core/src/main/java/org/springframework/core/io/buffer/PooledDataBuffer.java
View File

@@ -26,13 +26,13 @@ package org.springframework.core.io.buffer;
public interface PooledDataBuffer extends DataBuffer {
/**
* Increases the reference count for this buffer by one.
* Increase the reference count for this buffer by one.
* @return this buffer
*/
PooledDataBuffer retain();
/**
* Decreases the reference count for this buffer by one, and releases it
* Decrease the reference count for this buffer by one, and release it
* once the count reaches zero.
* @return {@code true} if the buffer was released; {@code false} otherwise.
*/

4
spring-core/src/main/java/org/springframework/core/io/buffer/support/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Support classes for Spring's byte buffer abstraction.
*/
package org.springframework.core.io.buffer.support;