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