Remove use of CompositeByteBuf in NettyDataBuffer

Prior to this commit, NettyDataBuffer had a optimization in
write(ByteBuf...), where it used a CompositeByteBuf to hold the original
and the parameter buffer.
Unfortunately, this procedure has nasty consequences when splicing
buffers (see https://stackoverflow.com/a/48111196/839733).

As of this commit, NettyDataBuffer stopped using CompositeByteBuf, and
 simply does ByteBuf.write().

Issue: SPR-16351

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

 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -25,11 +25,8 @@ import java.util.function.IntPredicate;
 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
@@ -42,7 +39,7 @@ public class NettyDataBuffer implements PooledDataBuffer {
 
 	private final NettyDataBufferFactory dataBufferFactory;
 
-	private ByteBuf byteBuf;
+	private final ByteBuf byteBuf;
 
 
 	/**
@@ -174,8 +171,10 @@ public class NettyDataBuffer implements PooledDataBuffer {
 
 	@Override
 	public NettyDataBuffer write(DataBuffer... buffers) {
-		if (!ObjectUtils.isEmpty(buffers)) {
-			if (buffers[0] instanceof NettyDataBuffer) {
+		Assert.notNull(buffers, "'buffers' must not be null");
+
+		if (buffers.length > 0) {
+			if (hasNettyDataBuffers(buffers)) {
 				ByteBuf[] nativeBuffers = Arrays.stream(buffers)
 						.map(b -> ((NettyDataBuffer) b).getNativeBuffer())
 						.toArray(ByteBuf[]::new);
@@ -191,12 +190,23 @@ public class NettyDataBuffer implements PooledDataBuffer {
 		return this;
 	}
 
+	private static boolean hasNettyDataBuffers(DataBuffer[] dataBuffers) {
+		for (DataBuffer dataBuffer : dataBuffers) {
+			if (!(dataBuffer instanceof NettyDataBuffer)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
 	@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);
+
+		for (ByteBuffer buffer : buffers) {
+			this.byteBuf.writeBytes(buffer);
+		}
+		return this;
 	}
 
 	/**
@@ -208,17 +218,8 @@ public class NettyDataBuffer implements PooledDataBuffer {
 	public NettyDataBuffer write(ByteBuf... byteBufs) {
 		Assert.notNull(byteBufs, "'byteBufs' must not be null");
 
-		if (this.byteBuf instanceof CompositeByteBuf) {
-			CompositeByteBuf composite = (CompositeByteBuf) this.byteBuf;
-			composite.addComponents(true, byteBufs);
-		}
-		else {
-			ByteBuf oldByteBuf = this.byteBuf;
-			CompositeByteBuf composite = oldByteBuf.alloc().compositeBuffer(byteBufs.length + 1);
-			composite.addComponent(true, oldByteBuf);
-			composite.addComponents(true, byteBufs);
-
-			this.byteBuf = composite;
+		for (ByteBuf byteBuf : byteBufs) {
+			this.byteBuf.writeBytes(byteBuf);
 		}
 		return this;
 	}

spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java

 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -303,7 +303,7 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
 
 		assertArrayEquals(new byte[]{'a', 'b', 'c', 'd'}, result);
 
-		release(buffer1);
+		release(buffer1, buffer2, buffer3);
 	}
 
 	@Test
@@ -461,5 +461,23 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
 		release(buffer);
 	}
 
+	@Test
+	public void spr16351() {
+		DataBuffer buffer = createDataBuffer(6);
+		byte[] bytes = {'a', 'b', 'c', 'd', 'e', 'f'};
+		buffer.write(bytes);
+		DataBuffer slice = buffer.slice(3, 3);
+		buffer.writePosition(3);
+		buffer.write(slice);
+
+		assertEquals(6, buffer.readableByteCount());
+		byte[] result = new byte[6];
+		buffer.read(result);
+
+		assertArrayEquals(bytes, result);
+
+		release(buffer);
+	}
+
 
 }
\ No newline at end of file