1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.testsuite.transport.socket;
17
18 import io.netty.bootstrap.Bootstrap;
19 import io.netty.bootstrap.ServerBootstrap;
20 import io.netty.buffer.ByteBuf;
21 import io.netty.buffer.ByteBufAllocator;
22 import io.netty.buffer.CompositeByteBuf;
23 import io.netty.channel.Channel;
24 import io.netty.channel.ChannelConfig;
25 import io.netty.channel.ChannelFutureListener;
26 import io.netty.channel.ChannelHandlerContext;
27 import io.netty.channel.ChannelInboundHandlerAdapter;
28 import io.netty.channel.ChannelInitializer;
29 import io.netty.channel.ChannelOption;
30 import io.netty.util.ReferenceCountUtil;
31 import org.junit.jupiter.api.Test;
32 import org.junit.jupiter.api.TestInfo;
33 import org.junit.jupiter.api.Timeout;
34
35 import java.io.IOException;
36 import java.util.concurrent.CountDownLatch;
37 import java.util.concurrent.ThreadLocalRandom;
38 import java.util.concurrent.TimeUnit;
39 import java.util.concurrent.atomic.AtomicReference;
40
41 import static org.junit.jupiter.api.Assertions.assertEquals;
42
43 public class CompositeBufferGatheringWriteTest extends AbstractSocketTest {
44 private static final int EXPECTED_BYTES = 20;
45
46 @Test
47 @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
48 public void testSingleCompositeBufferWrite(TestInfo testInfo) throws Throwable {
49 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
50 @Override
51 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
52 testSingleCompositeBufferWrite(serverBootstrap, bootstrap);
53 }
54 });
55 }
56
57 public void testSingleCompositeBufferWrite(ServerBootstrap sb, Bootstrap cb) throws Throwable {
58 Channel serverChannel = null;
59 Channel clientChannel = null;
60 try {
61 final CountDownLatch latch = new CountDownLatch(1);
62 final AtomicReference<Object> clientReceived = new AtomicReference<Object>();
63 sb.childHandler(new ChannelInitializer<Channel>() {
64 @Override
65 protected void initChannel(Channel ch) throws Exception {
66 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
67 @Override
68 public void channelActive(ChannelHandlerContext ctx) throws Exception {
69 ctx.writeAndFlush(newCompositeBuffer(ctx.alloc()))
70 .addListener(ChannelFutureListener.CLOSE);
71 }
72 });
73 }
74 });
75 cb.handler(new ChannelInitializer<Channel>() {
76 @Override
77 protected void initChannel(Channel ch) throws Exception {
78 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
79 private ByteBuf aggregator;
80 @Override
81 public void handlerAdded(ChannelHandlerContext ctx) {
82 aggregator = ctx.alloc().buffer(EXPECTED_BYTES);
83 }
84
85 @Override
86 public void channelRead(ChannelHandlerContext ctx, Object msg) {
87 try {
88 if (msg instanceof ByteBuf) {
89 aggregator.writeBytes((ByteBuf) msg);
90 }
91 } finally {
92 ReferenceCountUtil.release(msg);
93 }
94 }
95
96 @Override
97 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
98
99 if (!(cause instanceof IOException)) {
100 clientReceived.set(cause);
101 latch.countDown();
102 } else if (!cause.getMessage().contains("reset")) {
103 logger.warn("{} client got weird exception",
104 CompositeBufferGatheringWriteTest.this.getClass(), cause);
105 }
106 }
107
108 @Override
109 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
110 if (clientReceived.compareAndSet(null, aggregator)) {
111 try {
112 assertEquals(EXPECTED_BYTES, aggregator.readableBytes());
113 } catch (Throwable cause) {
114 aggregator.release();
115 aggregator = null;
116 clientReceived.set(cause);
117 } finally {
118 latch.countDown();
119 }
120 }
121 }
122 });
123 }
124 });
125
126 serverChannel = sb.bind().syncUninterruptibly().channel();
127 clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
128
129 ByteBuf expected = newCompositeBuffer(clientChannel.alloc());
130 latch.await();
131 Object received = clientReceived.get();
132 if (received instanceof ByteBuf) {
133 ByteBuf actual = (ByteBuf) received;
134 assertEquals(expected, actual);
135 expected.release();
136 actual.release();
137 } else {
138 expected.release();
139 throw (Throwable) received;
140 }
141 } finally {
142 if (clientChannel != null) {
143 clientChannel.close().sync();
144 }
145 if (serverChannel != null) {
146 serverChannel.close().sync();
147 }
148 }
149 }
150
151 @Test
152 @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
153 public void testCompositeBufferPartialWriteDoesNotCorruptData(TestInfo testInfo) throws Throwable {
154 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
155 @Override
156 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
157 testCompositeBufferPartialWriteDoesNotCorruptData(serverBootstrap, bootstrap);
158 }
159 });
160 }
161
162 protected void compositeBufferPartialWriteDoesNotCorruptDataInitServerConfig(ChannelConfig config,
163 int soSndBuf) {
164 }
165
166 public void testCompositeBufferPartialWriteDoesNotCorruptData(ServerBootstrap sb, Bootstrap cb) throws Throwable {
167
168
169
170
171 Channel serverChannel = null;
172 Channel clientChannel = null;
173 try {
174 final int soSndBuf = 1024;
175 ByteBufAllocator alloc = ByteBufAllocator.DEFAULT;
176 final ByteBuf expectedContent = alloc.buffer(soSndBuf * 2);
177 expectedContent.writeBytes(newRandomBytes(expectedContent.writableBytes()));
178 final CountDownLatch latch = new CountDownLatch(1);
179 final AtomicReference<Object> clientReceived = new AtomicReference<Object>();
180 sb.childOption(ChannelOption.SO_SNDBUF, soSndBuf)
181 .childHandler(new ChannelInitializer<Channel>() {
182 @Override
183 protected void initChannel(Channel ch) throws Exception {
184 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
185 @Override
186 public void channelActive(ChannelHandlerContext ctx) throws Exception {
187 compositeBufferPartialWriteDoesNotCorruptDataInitServerConfig(ctx.channel().config(),
188 soSndBuf);
189
190 int offset = soSndBuf - 100;
191 ctx.write(expectedContent.retainedSlice(expectedContent.readerIndex(), offset));
192
193
194 CompositeByteBuf compositeByteBuf = ctx.alloc().compositeBuffer();
195 compositeByteBuf.addComponent(true,
196 expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 50));
197 offset += 50;
198 compositeByteBuf.addComponent(true,
199 expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 200));
200 offset += 200;
201 ctx.write(compositeByteBuf);
202
203
204
205 ctx.write(expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 50));
206 offset += 50;
207
208
209 ctx.writeAndFlush(expectedContent.retainedSlice(expectedContent.readerIndex() + offset,
210 expectedContent.readableBytes() - expectedContent.readerIndex() - offset))
211 .addListener(ChannelFutureListener.CLOSE);
212 }
213
214 @Override
215 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
216
217 if (!(cause instanceof IOException)) {
218 clientReceived.set(cause);
219 latch.countDown();
220 }
221 }
222 });
223 }
224 });
225 cb.handler(new ChannelInitializer<Channel>() {
226 @Override
227 protected void initChannel(Channel ch) throws Exception {
228 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
229 private ByteBuf aggregator;
230 @Override
231 public void handlerAdded(ChannelHandlerContext ctx) {
232 aggregator = ctx.alloc().buffer(expectedContent.readableBytes());
233 }
234
235 @Override
236 public void channelRead(ChannelHandlerContext ctx, Object msg) {
237 try {
238 if (msg instanceof ByteBuf) {
239 aggregator.writeBytes((ByteBuf) msg);
240 }
241 } finally {
242 ReferenceCountUtil.release(msg);
243 }
244 }
245
246 @Override
247 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
248
249 if (!(cause instanceof IOException)) {
250 clientReceived.set(cause);
251 latch.countDown();
252 }
253 }
254
255 @Override
256 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
257 if (clientReceived.compareAndSet(null, aggregator)) {
258 try {
259 assertEquals(expectedContent.readableBytes(), aggregator.readableBytes());
260 } catch (Throwable cause) {
261 aggregator.release();
262 aggregator = null;
263 clientReceived.set(cause);
264 } finally {
265 latch.countDown();
266 }
267 }
268 }
269 });
270 }
271 });
272
273 serverChannel = sb.bind().syncUninterruptibly().channel();
274 clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
275
276 latch.await();
277 Object received = clientReceived.get();
278 if (received instanceof ByteBuf) {
279 ByteBuf actual = (ByteBuf) received;
280 assertEquals(expectedContent, actual);
281 expectedContent.release();
282 actual.release();
283 } else {
284 expectedContent.release();
285 throw (Throwable) received;
286 }
287 } finally {
288 if (clientChannel != null) {
289 clientChannel.close().sync();
290 }
291 if (serverChannel != null) {
292 serverChannel.close().sync();
293 }
294 }
295 }
296
297 private static ByteBuf newCompositeBuffer(ByteBufAllocator alloc) {
298 CompositeByteBuf compositeByteBuf = alloc.compositeBuffer();
299 compositeByteBuf.addComponent(true, alloc.directBuffer(4).writeInt(100));
300 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(123));
301 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(456));
302 assertEquals(EXPECTED_BYTES, compositeByteBuf.readableBytes());
303 return compositeByteBuf;
304 }
305
306 private static byte[] newRandomBytes(int size) {
307 byte[] bytes = new byte[size];
308 ThreadLocalRandom.current().nextBytes(bytes);
309 return bytes;
310 }
311 }