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.CompositeByteBuf;
22 import io.netty.channel.Channel;
23 import io.netty.channel.ChannelFuture;
24 import io.netty.channel.ChannelFutureListener;
25 import io.netty.channel.ChannelHandlerContext;
26 import io.netty.channel.ChannelInboundHandlerAdapter;
27 import io.netty.channel.ChannelInitializer;
28 import io.netty.channel.ChannelOption;
29 import io.netty.channel.ChannelPromise;
30 import io.netty.channel.socket.DuplexChannel;
31 import io.netty.channel.socket.SocketChannel;
32 import io.netty.util.concurrent.ImmediateEventExecutor;
33 import io.netty.util.concurrent.Promise;
34 import org.junit.jupiter.api.Test;
35 import org.junit.jupiter.api.TestInfo;
36 import org.junit.jupiter.api.Timeout;
37
38 import java.io.ByteArrayOutputStream;
39 import java.net.InetSocketAddress;
40 import java.net.SocketAddress;
41 import java.util.concurrent.BlockingQueue;
42 import java.util.concurrent.LinkedBlockingQueue;
43 import java.util.concurrent.Semaphore;
44 import java.util.concurrent.TimeUnit;
45 import java.util.concurrent.atomic.AtomicBoolean;
46 import java.util.function.Function;
47
48 import static io.netty.buffer.ByteBufUtil.writeAscii;
49 import static io.netty.buffer.UnpooledByteBufAllocator.DEFAULT;
50 import static io.netty.util.CharsetUtil.US_ASCII;
51 import static org.junit.jupiter.api.Assertions.assertEquals;
52 import static org.junit.jupiter.api.Assertions.assertFalse;
53 import static org.junit.jupiter.api.Assertions.assertNotNull;
54 import static org.junit.jupiter.api.Assertions.assertNull;
55 import static org.junit.jupiter.api.Assertions.assertTrue;
56
57 public class SocketConnectTest extends AbstractSocketTest {
58
59 @Test
60 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
61 public void testCloseTwice(TestInfo testInfo) throws Throwable {
62 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
63 @Override
64 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
65 testCloseTwice(serverBootstrap, bootstrap);
66 }
67 });
68 }
69
70 public void testCloseTwice(ServerBootstrap sb, Bootstrap cb) throws Throwable {
71 Channel serverChannel = null;
72 Channel clientChannel = null;
73 try {
74 serverChannel = sb.childHandler(new ChannelInboundHandlerAdapter()).bind().syncUninterruptibly().channel();
75 final BlockingQueue<ChannelFuture> futures = new LinkedBlockingQueue<>();
76 clientChannel = cb.handler(new ChannelInboundHandlerAdapter() {
77 @Override
78 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
79 futures.add(ctx.close());
80 }
81 })
82 .connect(serverChannel.localAddress()).syncUninterruptibly().channel();
83 clientChannel.pipeline().fireUserEventTriggered("test");
84 clientChannel.close().syncUninterruptibly();
85 futures.take().sync();
86 clientChannel = null;
87
88 serverChannel.close().syncUninterruptibly();
89 serverChannel.close().syncUninterruptibly();
90 serverChannel = null;
91 } finally {
92 if (clientChannel != null) {
93 clientChannel.close().syncUninterruptibly();
94 }
95 if (serverChannel != null) {
96 serverChannel.close().syncUninterruptibly();
97 }
98 }
99 }
100
101 @Test
102 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
103 public void testLocalAddressAfterConnect(TestInfo testInfo) throws Throwable {
104 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
105 @Override
106 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
107 testLocalAddressAfterConnect(serverBootstrap, bootstrap);
108 }
109 });
110 }
111
112 public void testLocalAddressAfterConnect(ServerBootstrap sb, Bootstrap cb) throws Throwable {
113 Channel serverChannel = null;
114 Channel clientChannel = null;
115 try {
116 final Promise<InetSocketAddress> localAddressPromise = ImmediateEventExecutor.INSTANCE.newPromise();
117 serverChannel = sb.childHandler(new ChannelInboundHandlerAdapter() {
118 @Override
119 public void channelActive(ChannelHandlerContext ctx) throws Exception {
120 localAddressPromise.setSuccess((InetSocketAddress) ctx.channel().localAddress());
121 }
122 }).bind().syncUninterruptibly().channel();
123
124 clientChannel = cb.handler(new ChannelInboundHandlerAdapter()).register().syncUninterruptibly().channel();
125
126 assertNull(clientChannel.localAddress());
127 assertNull(clientChannel.remoteAddress());
128
129 clientChannel.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
130 assertLocalAddress((InetSocketAddress) clientChannel.localAddress());
131 assertNotNull(clientChannel.remoteAddress());
132
133 assertLocalAddress(localAddressPromise.get());
134 } finally {
135 if (clientChannel != null) {
136 clientChannel.close().syncUninterruptibly();
137 }
138 if (serverChannel != null) {
139 serverChannel.close().syncUninterruptibly();
140 }
141 }
142 }
143
144 @Test
145 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
146 public void testChannelEventsFiredWhenClosedDirectly(TestInfo testInfo) throws Throwable {
147 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
148 @Override
149 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
150 testChannelEventsFiredWhenClosedDirectly(serverBootstrap, bootstrap);
151 }
152 });
153 }
154
155 public void testChannelEventsFiredWhenClosedDirectly(ServerBootstrap sb, Bootstrap cb) throws Throwable {
156 final BlockingQueue<Integer> events = new LinkedBlockingQueue<Integer>();
157
158 Channel sc = null;
159 Channel cc = null;
160 try {
161 sb.childHandler(new ChannelInboundHandlerAdapter());
162 sc = sb.bind().syncUninterruptibly().channel();
163
164 cb.handler(new ChannelInboundHandlerAdapter() {
165 @Override
166 public void channelActive(ChannelHandlerContext ctx) throws Exception {
167 events.add(0);
168 }
169
170 @Override
171 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
172 events.add(1);
173 }
174 });
175
176 cc = cb.connect(sc.localAddress()).addListener(ChannelFutureListener.CLOSE).
177 syncUninterruptibly().channel();
178 assertEquals(0, events.take().intValue());
179 assertEquals(1, events.take().intValue());
180 } finally {
181 if (cc != null) {
182 cc.close();
183 }
184 if (sc != null) {
185 sc.close();
186 }
187 }
188 }
189
190 @Test
191 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
192 public void testWriteWithFastOpenBeforeConnect(TestInfo testInfo) throws Throwable {
193 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
194 @Override
195 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
196 testWriteWithFastOpenBeforeConnect(serverBootstrap, bootstrap);
197 }
198 });
199 }
200
201 @Test
202 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
203 public void testWriteWithFastOpenBeforeConnectDirectBufferReaderIndex(TestInfo testInfo) throws Throwable {
204 run(testInfo, (serverBootstrap, bootstrap) ->
205 testWriteWithFastOpenBeforeConnect(serverBootstrap, bootstrap,
206 channel -> directBufferWithReaderIndex(
207 channel, "BAD-PREFIX-", "[fastopen-reader-index]"),
208 "[fastopen-reader-index]"));
209 }
210
211 @Test
212 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
213 public void testWriteWithFastOpenBeforeConnectCompositeBuffer(TestInfo testInfo) throws Throwable {
214 run(testInfo, (serverBootstrap, bootstrap) ->
215 testWriteWithFastOpenBeforeConnect(serverBootstrap, bootstrap,
216 channel -> compositeBuffer(channel, "[fastopen-", "composite-", "data]"),
217 "[fastopen-composite-data]"));
218 }
219
220 @Test
221 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
222 public void testWriteWithFastOpenBeforeConnectCompositeBufferReaderIndex(TestInfo testInfo) throws Throwable {
223 run(testInfo, (serverBootstrap, bootstrap) ->
224 testWriteWithFastOpenBeforeConnect(serverBootstrap, bootstrap,
225 channel -> compositeBufferWithReaderIndex(
226 channel, "BAD-PREFIX-", "[fastopen-", "composite-reader-", "index]"),
227 "[fastopen-composite-reader-index]"));
228 }
229
230 public void testWriteWithFastOpenBeforeConnect(ServerBootstrap sb, Bootstrap cb) throws Throwable {
231 testWriteWithFastOpenBeforeConnect(sb, cb, channel -> writeAscii(DEFAULT, "[fastopen]"), "[fastopen]");
232 }
233
234 protected void testWriteWithFastOpenBeforeConnect(ServerBootstrap sb, Bootstrap cb,
235 Function<Channel, ByteBuf> initialDataFactory,
236 String expectedInitialData) throws Throwable {
237 enableTcpFastOpen(sb, cb);
238 sb.childOption(ChannelOption.AUTO_READ, true);
239 cb.option(ChannelOption.AUTO_READ, true);
240
241 sb.childHandler(new ChannelInitializer<SocketChannel>() {
242 @Override
243 protected void initChannel(SocketChannel ch) throws Exception {
244 ch.pipeline().addLast(new EchoServerHandler());
245 }
246 });
247
248 Channel sc = sb.bind().sync().channel();
249 connectAndVerifyDataTransfer(cb, sc, initialDataFactory, expectedInitialData);
250 connectAndVerifyDataTransfer(cb, sc, initialDataFactory, expectedInitialData);
251 }
252
253 private static void connectAndVerifyDataTransfer(Bootstrap cb, Channel sc,
254 Function<Channel, ByteBuf> initialDataFactory,
255 String expectedInitialData)
256 throws InterruptedException {
257 BufferingClientHandler handler = new BufferingClientHandler();
258 cb.handler(handler);
259 ChannelFuture register = cb.register();
260 Channel channel = register.sync().channel();
261 ChannelFuture write = channel.write(initialDataFactory.apply(channel));
262 SocketAddress remoteAddress = sc.localAddress();
263 ChannelFuture connectFuture = channel.connect(remoteAddress);
264 Channel cc = connectFuture.sync().channel();
265 cc.writeAndFlush(writeAscii(DEFAULT, "[normal data]")).sync();
266 write.sync();
267 String expectedString = expectedInitialData + "[normal data]";
268 String result = handler.collectBuffer(expectedString.getBytes(US_ASCII).length);
269 cc.disconnect().sync();
270 assertEquals(expectedString, result);
271 }
272
273 @Test
274 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
275 public void testWriteFromFastOpenInitialDataListener(TestInfo testInfo) throws Throwable {
276 run(testInfo, this::testWriteFromFastOpenInitialDataListener);
277 }
278
279 public void testWriteFromFastOpenInitialDataListener(ServerBootstrap sb, Bootstrap cb) throws Throwable {
280 enableTcpFastOpen(sb, cb);
281 sb.childOption(ChannelOption.AUTO_READ, true);
282 cb.option(ChannelOption.AUTO_READ, true);
283
284 sb.childHandler(new ChannelInitializer<SocketChannel>() {
285 @Override
286 protected void initChannel(SocketChannel ch) throws Exception {
287 ch.pipeline().addLast(new EchoServerHandler());
288 }
289 });
290
291 Channel sc = sb.bind().sync().channel();
292
293
294 connectAndWriteFromInitialDataListener(cb, sc);
295 connectAndWriteFromInitialDataListener(cb, sc);
296 }
297
298 private static void connectAndWriteFromInitialDataListener(Bootstrap cb, Channel sc) throws Exception {
299 BufferingClientHandler handler = new BufferingClientHandler();
300 cb.handler(handler);
301 Channel channel = cb.register().sync().channel();
302 AtomicBoolean activeInListener = new AtomicBoolean();
303 BlockingQueue<ChannelFuture> followUpFutures = new LinkedBlockingQueue<>();
304 ChannelFuture initialFuture = channel.write(writeAscii(DEFAULT, "[fastopen]")).addListener(f -> {
305 activeInListener.set(channel.isActive());
306 followUpFutures.add(channel.writeAndFlush(writeAscii(DEFAULT, "[follow-up]")));
307 });
308 Channel cc = channel.connect(sc.localAddress()).sync().channel();
309
310 cc.flush();
311
312 initialFuture.sync();
313
314 ChannelFuture followUpFuture = followUpFutures.poll(2, TimeUnit.SECONDS);
315 assertNotNull(followUpFuture, "listener of the initial write was not notified");
316 assertTrue(activeInListener.get(), "channel was not active when the initial write was notified");
317 followUpFuture.sync();
318 String expectedString = "[fastopen][follow-up]";
319 String result = handler.collectBuffer(expectedString.getBytes(US_ASCII).length);
320 cc.close().sync();
321 assertEquals(expectedString, result);
322 }
323
324 @Test
325 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
326 public void testShutdownOutputFromFastOpenConnectListener(TestInfo testInfo) throws Throwable {
327 run(testInfo, this::testShutdownOutputFromFastOpenConnectListener);
328 }
329
330 public void testShutdownOutputFromFastOpenConnectListener(ServerBootstrap sb, Bootstrap cb) throws Throwable {
331 enableTcpFastOpen(sb, cb);
332 sb.childOption(ChannelOption.AUTO_READ, true);
333 BlockingQueue<String> received = new LinkedBlockingQueue<>();
334 sb.childHandler(new ChannelInitializer<SocketChannel>() {
335 @Override
336 protected void initChannel(SocketChannel ch) {
337 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
338 private final StringBuilder data = new StringBuilder();
339
340 @Override
341 public void channelRead(ChannelHandlerContext ctx, Object msg) {
342 ByteBuf buf = (ByteBuf) msg;
343 data.append(buf.toString(US_ASCII));
344 buf.release();
345 }
346
347 @Override
348 public void channelInactive(ChannelHandlerContext ctx) {
349 received.add(data.toString());
350 }
351 });
352 }
353 });
354
355 Channel sc = sb.bind().sync().channel();
356
357
358 shutdownOutputFromConnectListener(cb, sc, received);
359 shutdownOutputFromConnectListener(cb, sc, received);
360 }
361
362 private static void shutdownOutputFromConnectListener(Bootstrap cb, Channel sc, BlockingQueue<String> received)
363 throws Exception {
364 cb.handler(new ChannelInboundHandlerAdapter());
365 Channel channel = cb.register().sync().channel();
366 ChannelFuture initialFuture = channel.write(writeAscii(DEFAULT, "[fastopen]"));
367
368 ChannelPromise connectPromise = channel.newPromise();
369 connectPromise.addListener(f -> ((DuplexChannel) channel).shutdownOutput());
370 try {
371 channel.connect(sc.localAddress(), connectPromise).sync();
372 String data = received.poll(2, TimeUnit.SECONDS);
373 assertNotNull(data, "connection was not closed by the server");
374 initialFuture.await();
375
376 if (initialFuture.isSuccess()) {
377 assertEquals("[fastopen]", data);
378 } else {
379 assertEquals("", data, "initial data was sent but the write failed");
380 }
381 } finally {
382 channel.close().sync();
383 }
384 }
385
386 private static ByteBuf directBufferWithReaderIndex(Channel channel, String prefix, String data) {
387 ByteBuf buffer = directBuffer(channel, prefix + data);
388 buffer.readerIndex(prefix.getBytes(US_ASCII).length);
389 return buffer;
390 }
391
392 private static ByteBuf compositeBufferWithReaderIndex(Channel channel, String prefix, String... parts) {
393 CompositeByteBuf buffer = channel.alloc().compositeDirectBuffer(parts.length + 1);
394 buffer.addComponent(true, directBuffer(channel, prefix));
395 for (String part : parts) {
396 buffer.addComponent(true, directBuffer(channel, part));
397 }
398 buffer.readerIndex(prefix.getBytes(US_ASCII).length);
399 return buffer;
400 }
401
402 private static ByteBuf compositeBuffer(Channel channel, String... parts) {
403 CompositeByteBuf buffer = channel.alloc().compositeDirectBuffer(parts.length);
404 for (String part : parts) {
405 buffer.addComponent(true, directBuffer(channel, part));
406 }
407 return buffer;
408 }
409
410 private static ByteBuf directBuffer(Channel channel, String data) {
411 byte[] bytes = data.getBytes(US_ASCII);
412 ByteBuf buffer = channel.alloc().directBuffer(bytes.length);
413 buffer.writeBytes(bytes);
414 return buffer;
415 }
416
417 protected void enableTcpFastOpen(ServerBootstrap sb, Bootstrap cb) {
418
419 sb.option(ChannelOption.TCP_FASTOPEN, 5);
420 cb.option(ChannelOption.TCP_FASTOPEN_CONNECT, true);
421 }
422
423 private static void assertLocalAddress(InetSocketAddress address) {
424 assertTrue(address.getPort() > 0);
425 assertFalse(address.getAddress().isAnyLocalAddress());
426 }
427
428 private static class BufferingClientHandler extends ChannelInboundHandlerAdapter {
429 private final Semaphore semaphore = new Semaphore(0);
430 private final ByteArrayOutputStream streamBuffer = new ByteArrayOutputStream();
431
432 @Override
433 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
434 if (msg instanceof ByteBuf) {
435 ByteBuf buf = (ByteBuf) msg;
436 int readableBytes = buf.readableBytes();
437 buf.readBytes(streamBuffer, readableBytes);
438 semaphore.release(readableBytes);
439 buf.release();
440 } else {
441 throw new IllegalArgumentException("Unexpected message type: " + msg);
442 }
443 }
444
445 String collectBuffer(int expectedBytes) throws InterruptedException {
446 semaphore.acquire(expectedBytes);
447 byte[] bytes = streamBuffer.toByteArray();
448 streamBuffer.reset();
449 return new String(bytes, US_ASCII);
450 }
451 }
452
453 private static final class EchoServerHandler extends ChannelInboundHandlerAdapter {
454 @Override
455 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
456 if (msg instanceof ByteBuf) {
457 ByteBuf buffer = ctx.alloc().buffer();
458 ByteBuf buf = (ByteBuf) msg;
459 buffer.writeBytes(buf);
460 buf.release();
461 ctx.channel().writeAndFlush(buffer);
462 } else {
463 throw new IllegalArgumentException("Unexpected message type: " + msg);
464 }
465 }
466 }
467 }