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.buffer.ByteBuf;
20 import io.netty.channel.Channel;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.channel.ChannelHandlerContext;
23 import io.netty.channel.ChannelInboundHandlerAdapter;
24 import io.netty.channel.ChannelInitializer;
25 import io.netty.channel.ChannelOption;
26 import io.netty.channel.FixedRecvByteBufAllocator;
27 import io.netty.channel.socket.SocketChannel;
28 import io.netty.util.NetUtil;
29 import io.netty.util.ReferenceCountUtil;
30 import io.netty.util.concurrent.Future;
31 import io.netty.util.concurrent.ImmediateEventExecutor;
32 import io.netty.util.concurrent.Promise;
33 import org.junit.jupiter.api.Test;
34 import org.junit.jupiter.api.TestInfo;
35 import org.junit.jupiter.api.Timeout;
36
37 import java.io.IOException;
38 import java.io.OutputStream;
39 import java.net.InetSocketAddress;
40 import java.net.ServerSocket;
41 import java.net.Socket;
42 import java.net.SocketAddress;
43 import java.nio.channels.ClosedChannelException;
44 import java.security.MessageDigest;
45 import java.security.NoSuchAlgorithmException;
46 import java.util.concurrent.Callable;
47 import java.util.concurrent.TimeUnit;
48 import java.util.concurrent.atomic.AtomicReference;
49
50 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
51 import static org.junit.jupiter.api.Assertions.assertEquals;
52 import static org.junit.jupiter.api.Assertions.fail;
53
54 public class SocketChannelEOFTest extends AbstractClientSocketTest {
55
56 private static final long PAYLOAD_BYTES = 32L * 1024 * 1024;
57 private static final int CHUNK_BYTES = 8 * 1024;
58 private static final long PAUSE_AFTER_BYTES = 128L * 1024;
59 private static final long PAUSE_MILLIS = 750;
60 private static final byte[] CHUNK = createChunk();
61
62 @Test
63 @Timeout(30)
64 public void readAllPendingOnEOF(TestInfo info) throws Throwable {
65 run(info, new Runner<Bootstrap>() {
66 @Override
67 public void run(Bootstrap bootstrap) throws Throwable {
68 PayloadServer server = new PayloadServer(PAYLOAD_BYTES);
69 Channel ch = null;
70 try {
71 SocketAddress address = server.bindAndAccept();
72 final ReadHandler handler = new ReadHandler(PAYLOAD_BYTES, PAUSE_AFTER_BYTES, PAUSE_MILLIS);
73 bootstrap.option(ChannelOption.AUTO_READ, false)
74 .option(ChannelOption.TCP_NODELAY, true)
75 .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(CHUNK_BYTES))
76 .handler(new ChannelInitializer<SocketChannel>() {
77 @Override
78 protected void initChannel(SocketChannel channel) {
79 channel.pipeline().addLast(handler);
80 }
81 });
82 ch = bootstrap.connect(address).sync().channel();
83 handler.result().get(10, TimeUnit.SECONDS);
84 server.result().get(10, TimeUnit.SECONDS);
85 } finally {
86 server.close();
87 if (ch != null) {
88 ch.close().sync();
89 }
90 }
91 }
92 });
93 }
94
95 @Test
96 @Timeout(30)
97 public void readAllPendingOnEOFBeforeConnectComplete(TestInfo info) throws Throwable {
98 run(info, new Runner<Bootstrap>() {
99 @Override
100 public void run(Bootstrap bootstrap) throws Throwable {
101 final long payloadBytes = 2 * CHUNK_BYTES;
102 final PayloadServer server = new PayloadServer(payloadBytes);
103 Channel ch = null;
104 try {
105 final SocketAddress address = server.bindAndAccept();
106 final ReadHandler handler = new ReadHandler(payloadBytes, 0, 0);
107 bootstrap.option(ChannelOption.AUTO_READ, false)
108 .handler(new ChannelInitializer<SocketChannel>() {
109 @Override
110 protected void initChannel(SocketChannel channel) {
111 channel.pipeline().addLast(handler);
112 }
113 });
114
115 ch = bootstrap.register().sync().channel();
116 final Channel channel = ch;
117
118
119
120
121 ChannelFuture connectFuture = ch.eventLoop().submit(new Callable<ChannelFuture>() {
122 @Override
123 public ChannelFuture call() throws Exception {
124 ChannelFuture future = channel.connect(address);
125 server.result().await(10, TimeUnit.SECONDS);
126
127 Thread.sleep(100);
128 return future;
129 }
130 }).get(10, TimeUnit.SECONDS);
131 connectFuture.sync();
132 handler.result().get(10, TimeUnit.SECONDS);
133 server.result().get(10, TimeUnit.SECONDS);
134 } finally {
135 server.close();
136 if (ch != null) {
137 ch.close().sync();
138 }
139 }
140 }
141 });
142 }
143
144 private static byte[] createChunk() {
145 byte[] bytes = new byte[CHUNK_BYTES];
146 for (int index = 0; index < bytes.length; index++) {
147 bytes[index] = (byte) ((index * 31 + 7) & 0xff);
148 }
149 return bytes;
150 }
151
152 private static byte[] digest(long length) {
153 try {
154 MessageDigest digest = MessageDigest.getInstance("SHA-256");
155 for (long remaining = length; remaining > 0; remaining -= CHUNK.length) {
156 digest.update(CHUNK, 0, (int) Math.min(remaining, CHUNK.length));
157 }
158 return digest.digest();
159 } catch (NoSuchAlgorithmException e) {
160 throw new IllegalStateException(e);
161 }
162 }
163
164 private static final class ReadHandler extends ChannelInboundHandlerAdapter {
165
166 private final long expectedBytes;
167 private final byte[] expectedDigest;
168 private final long pauseAfterBytes;
169 private final long pauseMillis;
170 private final MessageDigest digest;
171 private final Promise<Void> result = ImmediateEventExecutor.INSTANCE.newPromise();
172 private Future<?> readFuture;
173 private long receivedBytes;
174 private Throwable failure;
175
176 private ReadHandler(long expectedBytes, long pauseAfterBytes, long pauseMillis) {
177 this.expectedBytes = expectedBytes;
178 this.expectedDigest = digest(expectedBytes);
179 this.pauseAfterBytes = pauseAfterBytes;
180 this.pauseMillis = pauseMillis;
181 try {
182 this.digest = MessageDigest.getInstance("SHA-256");
183 } catch (NoSuchAlgorithmException error) {
184 throw new IllegalStateException(error);
185 }
186 }
187
188 @Override
189 public void channelActive(ChannelHandlerContext context) {
190 requestRead(context);
191 }
192
193 @Override
194 public void channelRead(ChannelHandlerContext context, Object message) {
195 try {
196 ByteBuf bytes = (ByteBuf) message;
197 receivedBytes += bytes.readableBytes();
198 digest.update(bytes.nioBuffer());
199 } finally {
200 ReferenceCountUtil.release(message);
201 }
202 }
203
204 @Override
205 public void channelReadComplete(final ChannelHandlerContext context) {
206 if (receivedBytes < expectedBytes
207 && receivedBytes >= pauseAfterBytes
208 && pauseMillis > 0
209 && readFuture == null) {
210 readFuture = context.executor().schedule(new Runnable() {
211 @Override
212 public void run() {
213 requestRead(context);
214 }
215 }, pauseMillis, TimeUnit.MILLISECONDS);
216 } else {
217 requestRead(context);
218 }
219 }
220
221 @Override
222 public void channelInactive(ChannelHandlerContext context) {
223 try {
224 if (failure != null) {
225 fail(failure);
226 }
227 assertEquals(expectedBytes, receivedBytes);
228 assertArrayEquals(expectedDigest, digest.digest());
229 } catch (Throwable cause) {
230 result.setFailure(cause);
231 return;
232 }
233 result.setSuccess(null);
234 }
235
236 @Override
237 public void exceptionCaught(ChannelHandlerContext context, Throwable error) {
238 if (!(error instanceof ClosedChannelException)) {
239 failure = error;
240 }
241 context.close();
242 }
243
244 @Override
245 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
246 super.handlerRemoved(ctx);
247 if (readFuture != null) {
248 readFuture.cancel(true);
249 }
250 }
251
252 private void requestRead(ChannelHandlerContext context) {
253 if (context.channel().isActive()) {
254 context.read();
255 }
256 }
257
258 Future<Void> result() {
259 return result;
260 }
261 }
262
263 private static final class PayloadServer implements AutoCloseable {
264 private final long sendBytes;
265 private final ServerSocket serverSocket;
266 private final Promise<Void> result = ImmediateEventExecutor.INSTANCE.newPromise();
267 private final AtomicReference<Socket> accepted = new AtomicReference<Socket>();
268 private Thread thread;
269
270 PayloadServer(long sendBytes) throws IOException {
271 this.sendBytes = sendBytes;
272 this.serverSocket = new ServerSocket();
273 serverSocket.bind(new InetSocketAddress(NetUtil.LOCALHOST, 0), 1);
274 }
275
276 SocketAddress bindAndAccept() {
277 thread = new Thread(new Runnable() {
278 @Override
279 public void run() {
280 long sent = 0;
281 Throwable failure = null;
282 try {
283 Socket socket = serverSocket.accept();
284 accepted.set(socket);
285 socket.setTcpNoDelay(true);
286 OutputStream output = socket.getOutputStream();
287 for (long remaining = sendBytes; remaining > 0; remaining -= CHUNK.length) {
288 int length = (int) Math.min(remaining, CHUNK.length);
289 output.write(CHUNK, 0, length);
290 sent += length;
291 }
292 output.flush();
293 socket.shutdownOutput();
294 } catch (Throwable error) {
295 failure = error;
296 }
297 try {
298 if (failure != null) {
299 fail(failure);
300 }
301 assertEquals(sendBytes, sent);
302 } catch (Throwable error) {
303 result.setFailure(error);
304 return;
305 }
306 result.setSuccess(null);
307 }
308 }, "payload-server");
309 thread.setDaemon(true);
310 thread.start();
311 return serverSocket.getLocalSocketAddress();
312 }
313
314 Future<Void> result() {
315 return result;
316 }
317
318 @Override
319 public void close() throws Exception {
320 serverSocket.close();
321 Socket socket = accepted.get();
322 if (socket != null) {
323 socket.close();
324 }
325 if (thread != null) {
326 thread.join(TimeUnit.SECONDS.toMillis(5));
327 }
328 }
329 }
330 }