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