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 } else if (!cause.getMessage().contains("reset")) {
221 logger.warn("{} server got weird exception",
222 CompositeBufferGatheringWriteTest.this.getClass(), cause);
223 }
224 }
225 });
226 }
227 });
228 cb.handler(new ChannelInitializer<Channel>() {
229 @Override
230 protected void initChannel(Channel ch) throws Exception {
231 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
232 private ByteBuf aggregator;
233 @Override
234 public void handlerAdded(ChannelHandlerContext ctx) {
235 aggregator = ctx.alloc().buffer(expectedContent.readableBytes());
236 }
237
238 @Override
239 public void channelRead(ChannelHandlerContext ctx, Object msg) {
240 try {
241 if (msg instanceof ByteBuf) {
242 aggregator.writeBytes((ByteBuf) msg);
243 }
244 } finally {
245 ReferenceCountUtil.release(msg);
246 }
247 }
248
249 @Override
250 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
251
252 if (!(cause instanceof IOException)) {
253 clientReceived.set(cause);
254 latch.countDown();
255 } else if (!cause.getMessage().contains("reset")) {
256 logger.warn("{} client got weird exception",
257 CompositeBufferGatheringWriteTest.this.getClass(), cause);
258 }
259 }
260
261 @Override
262 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
263 if (clientReceived.compareAndSet(null, aggregator)) {
264 try {
265 assertEquals(expectedContent.readableBytes(), aggregator.readableBytes());
266 } catch (Throwable cause) {
267 aggregator.release();
268 aggregator = null;
269 clientReceived.set(cause);
270 } finally {
271 latch.countDown();
272 }
273 }
274 }
275 });
276 }
277 });
278
279 serverChannel = sb.bind().syncUninterruptibly().channel();
280 clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
281
282 latch.await();
283 Object received = clientReceived.get();
284 if (received instanceof ByteBuf) {
285 ByteBuf actual = (ByteBuf) received;
286 assertEquals(expectedContent, actual);
287 expectedContent.release();
288 actual.release();
289 } else {
290 expectedContent.release();
291 throw (Throwable) received;
292 }
293 } finally {
294 if (clientChannel != null) {
295 clientChannel.close().sync();
296 }
297 if (serverChannel != null) {
298 serverChannel.close().sync();
299 }
300 }
301 }
302
303 private static ByteBuf newCompositeBuffer(ByteBufAllocator alloc) {
304 CompositeByteBuf compositeByteBuf = alloc.compositeBuffer();
305 compositeByteBuf.addComponent(true, alloc.directBuffer(4).writeInt(100));
306 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(123));
307 compositeByteBuf.addComponent(true, alloc.directBuffer(8).writeLong(456));
308 assertEquals(EXPECTED_BYTES, compositeByteBuf.readableBytes());
309 return compositeByteBuf;
310 }
311
312 private static byte[] newRandomBytes(int size) {
313 byte[] bytes = new byte[size];
314 ThreadLocalRandom.current().nextBytes(bytes);
315 return bytes;
316 }
317 }