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.channel.Channel;
22 import io.netty.channel.ChannelHandlerContext;
23 import io.netty.channel.ChannelInitializer;
24 import io.netty.channel.ChannelOption;
25 import io.netty.channel.SimpleChannelInboundHandler;
26 import io.netty.handler.codec.FixedLengthFrameDecoder;
27 import io.netty.util.internal.PlatformDependent;
28 import org.junit.jupiter.api.Test;
29 import org.junit.jupiter.api.TestInfo;
30
31 import java.io.IOException;
32 import java.util.Random;
33 import java.util.SplittableRandom;
34 import java.util.concurrent.atomic.AtomicReference;
35
36 import static io.netty.testsuite.transport.TestsuitePermutation.randomBufferType;
37 import static org.junit.jupiter.api.Assertions.assertEquals;
38
39 public class SocketFixedLengthEchoTest extends AbstractSocketTest {
40
41 private static final Random random = new Random();
42 static final byte[] data = new byte[1048576];
43
44 static {
45 PlatformDependent.splittableRandomNextBytes(new SplittableRandom(random.nextLong()), data);
46 }
47
48 @Test
49 public void testFixedLengthEcho(TestInfo testInfo) throws Throwable {
50 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
51 @Override
52 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
53 testFixedLengthEcho(serverBootstrap, bootstrap);
54 }
55 });
56 }
57
58 @Test
59 public void testFixedLengthEchoNotAutoRead(TestInfo testInfo) throws Throwable {
60 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
61 @Override
62 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
63 testFixedLengthEchoNotAutoRead(serverBootstrap, bootstrap);
64 }
65 });
66 }
67
68 public void testFixedLengthEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
69 testFixedLengthEcho(sb, cb, true);
70 }
71
72 public void testFixedLengthEchoNotAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
73 testFixedLengthEcho(sb, cb, false);
74 }
75
76 private static void testFixedLengthEcho(ServerBootstrap sb, Bootstrap cb, boolean autoRead) throws Throwable {
77 final EchoHandler sh = new EchoHandler(autoRead);
78 final EchoHandler ch = new EchoHandler(autoRead);
79
80 sb.childOption(ChannelOption.AUTO_READ, autoRead);
81 sb.childHandler(new ChannelInitializer<Channel>() {
82 @Override
83 public void initChannel(Channel sch) throws Exception {
84 sch.pipeline().addLast("decoder", new FixedLengthFrameDecoder(1024));
85 sch.pipeline().addAfter("decoder", "handler", sh);
86 }
87 });
88
89 cb.option(ChannelOption.AUTO_READ, autoRead);
90 cb.handler(new ChannelInitializer<Channel>() {
91 @Override
92 public void initChannel(Channel sch) throws Exception {
93 sch.pipeline().addLast("decoder", new FixedLengthFrameDecoder(1024));
94 sch.pipeline().addAfter("decoder", "handler", ch);
95 }
96 });
97
98 Channel sc = sb.bind().sync().channel();
99 Channel cc = cb.connect(sc.localAddress()).sync().channel();
100 SplittableRandom rng = new SplittableRandom(random.nextLong());
101 for (int i = 0; i < data.length;) {
102 int length = Math.min(rng.nextInt(1024 * 3), data.length - i);
103 cc.writeAndFlush(randomBufferType(cc.alloc(), data, i, length));
104 i += length;
105 }
106
107 while (ch.counter < data.length) {
108 if (sh.exception.get() != null) {
109 break;
110 }
111 if (ch.exception.get() != null) {
112 break;
113 }
114
115 Thread.sleep(50);
116 }
117
118 while (sh.counter < data.length) {
119 if (sh.exception.get() != null) {
120 break;
121 }
122 if (ch.exception.get() != null) {
123 break;
124 }
125
126 Thread.sleep(50);
127 }
128
129 sh.channel.close().sync();
130 ch.channel.close().sync();
131 sc.close().sync();
132
133 if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
134 throw sh.exception.get();
135 }
136 if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
137 throw ch.exception.get();
138 }
139 if (sh.exception.get() != null) {
140 throw sh.exception.get();
141 }
142 if (ch.exception.get() != null) {
143 throw ch.exception.get();
144 }
145 }
146
147 private static class EchoHandler extends SimpleChannelInboundHandler<ByteBuf> {
148 private final boolean autoRead;
149 volatile Channel channel;
150 final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
151 volatile int counter;
152
153 EchoHandler(boolean autoRead) {
154 this.autoRead = autoRead;
155 }
156
157 @Override
158 public void channelActive(ChannelHandlerContext ctx) throws Exception {
159 channel = ctx.channel();
160 if (!autoRead) {
161 ctx.read();
162 }
163 }
164
165 @Override
166 public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
167 assertEquals(1024, msg.readableBytes());
168
169 byte[] actual = new byte[msg.readableBytes()];
170 msg.getBytes(msg.readerIndex(), actual);
171
172 int lastIdx = counter;
173 for (int i = 0; i < actual.length; i ++) {
174 assertEquals(data[i + lastIdx], actual[i]);
175 }
176
177
178
179 counter += actual.length;
180
181 if (channel.parent() != null) {
182 channel.write(msg.retain());
183 }
184 }
185
186 @Override
187 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
188 try {
189 ctx.flush();
190 } finally {
191 if (!autoRead) {
192 ctx.read();
193 }
194 }
195 }
196
197 @Override
198 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
199 if (exception.compareAndSet(null, cause)) {
200 ctx.close();
201 }
202 }
203 }
204 }