1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.testsuite.transport.sctp;
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.SimpleChannelInboundHandler;
25 import io.netty.channel.sctp.SctpChannel;
26 import io.netty.handler.codec.sctp.SctpInboundByteStreamHandler;
27 import io.netty.handler.codec.sctp.SctpMessageCompletionHandler;
28 import io.netty.handler.codec.sctp.SctpOutboundByteStreamHandler;
29 import io.netty.testsuite.util.TestUtils;
30 import io.netty.util.internal.PlatformDependent;
31 import org.junit.jupiter.api.Test;
32 import org.junit.jupiter.api.TestInfo;
33
34 import java.io.IOException;
35 import java.util.Random;
36 import java.util.SplittableRandom;
37 import java.util.concurrent.atomic.AtomicReference;
38
39 import static io.netty.testsuite.transport.TestsuitePermutation.randomBufferType;
40 import static org.junit.jupiter.api.Assertions.assertEquals;
41 import static org.junit.jupiter.api.Assumptions.assumeTrue;
42
43 public class SctpEchoTest extends AbstractSctpTest {
44
45 private static final Random random = new Random();
46 static final byte[] data = new byte[4096];
47
48 static {
49 PlatformDependent.splittableRandomNextBytes(new SplittableRandom(random.nextLong()), data);
50 }
51
52 @Test
53 public void testSimpleEcho(TestInfo testInfo) throws Throwable {
54 assumeTrue(TestUtils.isSctpSupported());
55 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
56 @Override
57 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
58 testSimpleEcho(serverBootstrap, bootstrap);
59 }
60 });
61 }
62
63 public void testSimpleEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
64 testSimpleEcho0(sb, cb, false);
65 }
66
67 @Test
68 public void testSimpleEchoUnordered(TestInfo testInfo) throws Throwable {
69 assumeTrue(TestUtils.isSctpSupported());
70 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
71 @Override
72 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
73 testSimpleEchoUnordered(serverBootstrap, bootstrap);
74 }
75 });
76 }
77
78 public void testSimpleEchoUnordered(ServerBootstrap sb, Bootstrap cb) throws Throwable {
79 testSimpleEcho0(sb, cb, true);
80 }
81
82 private static void testSimpleEcho0(ServerBootstrap sb, Bootstrap cb, final boolean unordered) throws Throwable {
83 final EchoHandler sh = new EchoHandler();
84 final EchoHandler ch = new EchoHandler();
85
86 sb.childHandler(new ChannelInitializer<SctpChannel>() {
87 @Override
88 public void initChannel(SctpChannel c) throws Exception {
89 c.pipeline().addLast(
90 new SctpMessageCompletionHandler(),
91 new SctpInboundByteStreamHandler(0, 0),
92 new SctpOutboundByteStreamHandler(0, 0, unordered),
93 sh);
94 }
95 });
96 cb.handler(new ChannelInitializer<SctpChannel>() {
97 @Override
98 public void initChannel(SctpChannel c) throws Exception {
99 c.pipeline().addLast(
100 new SctpMessageCompletionHandler(),
101 new SctpInboundByteStreamHandler(0, 0),
102 new SctpOutboundByteStreamHandler(0, 0, unordered),
103 ch);
104 }
105 });
106
107 Channel sc = sb.bind().sync().channel();
108 Channel cc = cb.connect(sc.localAddress()).sync().channel();
109
110 for (int i = 0; i < data.length;) {
111 int length = Math.min(random.nextInt(1024 * 64), data.length - i);
112 ByteBuf msg = randomBufferType(sc.alloc(), data, i, length);
113 cc.writeAndFlush(msg);
114 i += length;
115 }
116
117 while (ch.counter < data.length) {
118 if (sh.exception.get() != null) {
119 break;
120 }
121 if (ch.exception.get() != null) {
122 break;
123 }
124
125 Thread.sleep(50);
126 }
127
128 while (sh.counter < data.length) {
129 if (sh.exception.get() != null) {
130 break;
131 }
132 if (ch.exception.get() != null) {
133 break;
134 }
135
136 Thread.sleep(50);
137 }
138
139 sh.channel.close().sync();
140 ch.channel.close().sync();
141 sc.close().sync();
142
143 if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
144 throw sh.exception.get();
145 }
146 if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
147 throw ch.exception.get();
148 }
149 if (sh.exception.get() != null) {
150 throw sh.exception.get();
151 }
152 if (ch.exception.get() != null) {
153 throw ch.exception.get();
154 }
155 }
156
157 private static class EchoHandler extends SimpleChannelInboundHandler<ByteBuf> {
158 volatile Channel channel;
159 final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
160 volatile int counter;
161
162 @Override
163 public void channelActive(ChannelHandlerContext ctx) throws Exception {
164 channel = ctx.channel();
165 }
166
167 @Override
168 public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
169 byte[] actual = new byte[in.readableBytes()];
170 in.readBytes(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.writeAndFlush(randomBufferType(channel.alloc(), actual, 0, actual.length));
183 }
184 }
185
186 @Override
187 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
188 if (exception.compareAndSet(null, cause)) {
189 ctx.close();
190 }
191 }
192 }
193 }