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.concurrent.CountDownLatch;
38 import java.util.concurrent.ThreadLocalRandom;
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 final int soSndBuf = 1024;
202 ByteBufAllocator alloc = ByteBufAllocator.DEFAULT;
203 final ByteBuf expectedContent = alloc.buffer(soSndBuf * 2);
204 expectedContent.writeBytes(newRandomBytes(expectedContent.writableBytes()));
205 final CountDownLatch latch = new CountDownLatch(1);
206 final AtomicReference<Object> clientReceived = new AtomicReference<Object>();
207 sb.childOption(ChannelOption.SO_SNDBUF, soSndBuf)
208 .childHandler(new ChannelInitializer<Channel>() {
209 @Override
210 protected void initChannel(Channel ch) throws Exception {
211 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
212 @Override
213 public void channelActive(final ChannelHandlerContext ctx) throws Exception {
214 compositeBufferPartialWriteDoesNotCorruptDataInitServerConfig(ctx.channel().config(),
215 soSndBuf);
216
217 int offset = soSndBuf - 100;
218 ctx.write(expectedContent.retainedSlice(expectedContent.readerIndex(), offset));
219
220
221 CompositeByteBuf compositeByteBuf = ctx.alloc().compositeBuffer();
222 compositeByteBuf.addComponent(true,
223 expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 50));
224 offset += 50;
225 compositeByteBuf.addComponent(true,
226 expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 200));
227 offset += 200;
228 ctx.write(compositeByteBuf);
229
230
231
232 ctx.write(expectedContent.retainedSlice(expectedContent.readerIndex() + offset, 50));
233 offset += 50;
234
235
236 ctx.writeAndFlush(expectedContent.retainedSlice(expectedContent.readerIndex() + offset,
237 expectedContent.readableBytes() - expectedContent.readerIndex() - offset))
238 .addListener(new ChannelFutureListener() {
239 @Override
240 public void operationComplete(ChannelFuture future) {
241 if (!future.isSuccess()) {
242 ctx.close();
243 }
244
245
246
247 }
248 });
249 }
250
251 @Override
252 public void channelRead(ChannelHandlerContext ctx, Object msg) {
253
254 ReferenceCountUtil.release(msg);
255 ctx.close();
256 }
257
258 @Override
259 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
260
261 if (!(cause instanceof IOException)) {
262 clientReceived.set(cause);
263 latch.countDown();
264 } else if (!cause.getMessage().contains("reset")) {
265 logger.warn("{} server got weird exception",
266 CompositeBufferGatheringWriteTest.this.getClass(), cause);
267 }
268 }
269 });
270 }
271 });
272 cb.handler(new ChannelInitializer<Channel>() {
273 @Override
274 protected void initChannel(Channel ch) throws Exception {
275 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
276 private ByteBuf aggregator;
277 private boolean acked;
278 @Override
279 public void handlerAdded(ChannelHandlerContext ctx) {
280 aggregator = ctx.alloc().buffer(expectedContent.readableBytes());
281 }
282
283 @Override
284 public void channelRead(ChannelHandlerContext ctx, Object msg) {
285 try {
286 if (msg instanceof ByteBuf) {
287 aggregator.writeBytes((ByteBuf) msg);
288 if (!acked && aggregator.readableBytes() >= expectedContent.readableBytes()) {
289 acked = true;
290 if (clientReceived.compareAndSet(null, aggregator)) {
291 latch.countDown();
292 }
293
294 ctx.writeAndFlush(ctx.alloc().buffer(1).writeByte(0));
295 }
296 }
297 } finally {
298 ReferenceCountUtil.release(msg);
299 }
300 }
301
302 @Override
303 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
304
305 if (!(cause instanceof IOException)) {
306 clientReceived.set(cause);
307 latch.countDown();
308 } else if (!cause.getMessage().contains("reset")) {
309 logger.warn("{} client got weird exception",
310 CompositeBufferGatheringWriteTest.this.getClass(), cause);
311 }
312 }
313
314 @Override
315 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
316 if (clientReceived.compareAndSet(null, aggregator)) {
317 try {
318 assertEquals(expectedContent.readableBytes(), aggregator.readableBytes());
319 } catch (Throwable cause) {
320 aggregator.release();
321 aggregator = null;
322 clientReceived.set(cause);
323 } finally {
324 latch.countDown();
325 }
326 }
327 }
328 });
329 }
330 });
331
332 serverChannel = sb.bind().syncUninterruptibly().channel();
333 clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
334
335 latch.await();
336 Object received = clientReceived.get();
337 if (received instanceof ByteBuf) {
338 ByteBuf actual = (ByteBuf) received;
339 assertEquals(expectedContent, actual);
340 expectedContent.release();
341 actual.release();
342 } else {
343 expectedContent.release();
344 throw (Throwable) received;
345 }
346 } finally {
347 if (clientChannel != null) {
348 clientChannel.close().sync();
349 }
350 if (serverChannel != null) {
351 serverChannel.close().sync();
352 }
353 }
354 }
355
356 private static ByteBuf newCompositeBuffer(ByteBufAllocator alloc) {
357 CompositeByteBuf compositeByteBuf = alloc.compositeBuffer();
358 compositeByteBuf.addComponent(true, alloc.directBuffer(4).writeInt(100));
359 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(123));
360 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(456));
361 assertEquals(EXPECTED_BYTES, compositeByteBuf.readableBytes());
362 return compositeByteBuf;
363 }
364
365 private static byte[] newRandomBytes(int size) {
366 byte[] bytes = new byte[size];
367 ThreadLocalRandom.current().nextBytes(bytes);
368 return bytes;
369 }
370 }