1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty5.testsuite.transport.socket;
17
18 import io.netty5.bootstrap.Bootstrap;
19 import io.netty5.bootstrap.ServerBootstrap;
20 import io.netty5.channel.Channel;
21 import io.netty5.channel.ChannelHandlerContext;
22 import io.netty5.channel.ChannelInitializer;
23 import io.netty5.channel.ChannelOption;
24 import io.netty5.channel.SimpleChannelInboundHandler;
25 import io.netty5.handler.codec.DelimiterBasedFrameDecoder;
26 import io.netty5.handler.codec.Delimiters;
27 import io.netty5.handler.codec.string.StringDecoder;
28 import io.netty5.handler.codec.string.StringEncoder;
29 import io.netty5.util.CharsetUtil;
30 import io.netty5.util.concurrent.ImmediateEventExecutor;
31 import io.netty5.util.concurrent.Promise;
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.TimeUnit;
39 import java.util.concurrent.atomic.AtomicReference;
40
41 public class SocketStringEchoTest extends AbstractSocketTest {
42
43 static final Random random = new Random();
44 static final String[] data = new String[1024];
45
46 static {
47 for (int i = 0; i < data.length; i ++) {
48 int eLen = random.nextInt(512);
49 char[] e = new char[eLen];
50 for (int j = 0; j < eLen; j ++) {
51 e[j] = (char) ('a' + random.nextInt(26));
52 }
53
54 data[i] = new String(e);
55 }
56 }
57
58 @Test
59 @Timeout(value = 60000, unit = TimeUnit.MILLISECONDS)
60 public void testStringEcho(TestInfo testInfo) throws Throwable {
61 run(testInfo, this::testStringEcho);
62 }
63
64 public void testStringEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
65 testStringEcho(sb, cb, true);
66 }
67
68 @Test
69 @Timeout(value = 60000, unit = TimeUnit.MILLISECONDS)
70 public void testStringEchoNotAutoRead(TestInfo testInfo) throws Throwable {
71 run(testInfo, this::testStringEchoNotAutoRead);
72 }
73
74 public void testStringEchoNotAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
75 testStringEcho(sb, cb, false);
76 }
77
78 private static void testStringEcho(ServerBootstrap sb, Bootstrap cb, boolean autoRead) throws Throwable {
79 sb.childOption(ChannelOption.AUTO_READ, autoRead);
80 cb.option(ChannelOption.AUTO_READ, autoRead);
81
82 Promise<Void> serverDonePromise = ImmediateEventExecutor.INSTANCE.newPromise();
83 Promise<Void> clientDonePromise = ImmediateEventExecutor.INSTANCE.newPromise();
84 final StringEchoHandler sh = new StringEchoHandler(autoRead, serverDonePromise);
85 final StringEchoHandler ch = new StringEchoHandler(autoRead, clientDonePromise);
86
87 sb.childHandler(new ChannelInitializer<>() {
88 @Override
89 public void initChannel(Channel sch) {
90 sch.pipeline().addLast("framer", new DelimiterBasedFrameDecoder(512, Delimiters.lineDelimiter()));
91 sch.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.ISO_8859_1));
92 sch.pipeline().addBefore("decoder", "encoder", new StringEncoder(CharsetUtil.ISO_8859_1));
93 sch.pipeline().addAfter("decoder", "handler", sh);
94 }
95 });
96
97 cb.handler(new ChannelInitializer<>() {
98 @Override
99 public void initChannel(Channel sch) {
100 sch.pipeline().addLast("framer", new DelimiterBasedFrameDecoder(512, Delimiters.lineDelimiter()));
101 sch.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.ISO_8859_1));
102 sch.pipeline().addBefore("decoder", "encoder", new StringEncoder(CharsetUtil.ISO_8859_1));
103 sch.pipeline().addAfter("decoder", "handler", ch);
104 }
105 });
106
107 Channel sc = sb.bind().asStage().get();
108 Channel cc = cb.connect(sc.localAddress()).asStage().get();
109 for (String element : data) {
110 String delimiter = random.nextBoolean() ? "\r\n" : "\n";
111 cc.writeAndFlush(element + delimiter);
112 }
113
114 ch.donePromise.asFuture().asStage().sync();
115 sh.donePromise.asFuture().asStage().sync();
116 sh.channel.close().asStage().sync();
117 ch.channel.close().asStage().sync();
118 sc.close().asStage().sync();
119
120 if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
121 throw sh.exception.get();
122 }
123 if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
124 throw ch.exception.get();
125 }
126 if (sh.exception.get() != null) {
127 throw sh.exception.get();
128 }
129 if (ch.exception.get() != null) {
130 throw ch.exception.get();
131 }
132 }
133
134 static class StringEchoHandler extends SimpleChannelInboundHandler<String> {
135 private final boolean autoRead;
136 private final Promise<Void> donePromise;
137 private int dataIndex;
138 volatile Channel channel;
139 final AtomicReference<Throwable> exception = new AtomicReference<>();
140
141 StringEchoHandler(boolean autoRead, Promise<Void> donePromise) {
142 this.autoRead = autoRead;
143 this.donePromise = donePromise;
144 }
145
146 @Override
147 public void channelActive(ChannelHandlerContext ctx) throws Exception {
148 channel = ctx.channel();
149 if (!autoRead) {
150 ctx.read();
151 }
152 }
153
154 @Override
155 public void messageReceived(ChannelHandlerContext ctx, String msg) throws Exception {
156 if (!data[dataIndex].equals(msg)) {
157 donePromise.tryFailure(new IllegalStateException("index: " + dataIndex + " didn't match!"));
158 ctx.close();
159 return;
160 }
161
162 if (channel.parent() != null) {
163 String delimiter = random.nextBoolean() ? "\r\n" : "\n";
164 channel.write(msg + delimiter);
165 }
166
167 if (++dataIndex >= data.length) {
168 donePromise.setSuccess(null);
169 }
170 }
171
172 @Override
173 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
174 try {
175 ctx.flush();
176 } finally {
177 if (!autoRead) {
178 ctx.read();
179 }
180 }
181 }
182
183 @Override
184 public void channelExceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
185 if (exception.compareAndSet(null, cause)) {
186 donePromise.tryFailure(new IllegalStateException("exceptionCaught: " + ctx.channel(), cause));
187 ctx.close();
188 }
189 }
190
191 @Override
192 public void channelInactive(ChannelHandlerContext ctx) {
193 donePromise.tryFailure(new IllegalStateException("channelInactive: " + ctx.channel()));
194 }
195 }
196 }