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.buffer.Unpooled;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.channel.ChannelHandlerContext;
23 import io.netty.channel.ChannelInboundHandlerAdapter;
24 import io.netty.channel.ChannelOption;
25 import io.netty.channel.SimpleChannelInboundHandler;
26 import io.netty.channel.WriteBufferWaterMark;
27 import io.netty.channel.socket.SocketChannel;
28 import io.netty.channel.socket.oio.OioSocketChannel;
29 import org.junit.jupiter.api.Disabled;
30 import org.junit.jupiter.api.Test;
31 import org.junit.jupiter.api.TestInfo;
32 import org.junit.jupiter.api.Timeout;
33 import org.junit.jupiter.api.function.Executable;
34
35 import java.net.ServerSocket;
36 import java.net.Socket;
37 import java.net.SocketException;
38 import java.nio.channels.ClosedChannelException;
39 import java.util.concurrent.BlockingDeque;
40 import java.util.concurrent.BlockingQueue;
41 import java.util.concurrent.LinkedBlockingDeque;
42 import java.util.concurrent.LinkedBlockingQueue;
43 import java.util.concurrent.TimeUnit;
44
45 import static io.netty.testsuite.transport.TestsuitePermutation.randomBufferType;
46 import static org.junit.jupiter.api.Assertions.assertEquals;
47 import static org.junit.jupiter.api.Assertions.assertFalse;
48 import static org.junit.jupiter.api.Assertions.assertNull;
49 import static org.junit.jupiter.api.Assertions.assertThrows;
50 import static org.junit.jupiter.api.Assertions.assertTrue;
51 import static org.junit.jupiter.api.Assumptions.assumeFalse;
52
53 public class SocketShutdownOutputBySelfTest extends AbstractClientSocketTest {
54
55 @Test
56 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
57 public void testShutdownOutput(TestInfo testInfo) throws Throwable {
58 run(testInfo, new Runner<Bootstrap>() {
59 @Override
60 public void run(Bootstrap bootstrap) throws Throwable {
61 testShutdownOutput(bootstrap);
62 }
63 });
64 }
65
66 public void testShutdownOutput(Bootstrap cb) throws Throwable {
67 TestHandler h = new TestHandler();
68 ServerSocket ss = new ServerSocket();
69 Socket s = null;
70 SocketChannel ch = null;
71 try {
72 ss.bind(newSocketAddress());
73 ch = (SocketChannel) cb.handler(h).connect(ss.getLocalSocketAddress()).sync().channel();
74 assertTrue(ch.isActive());
75 assertFalse(ch.isOutputShutdown());
76
77 s = ss.accept();
78 ch.writeAndFlush(randomBufferType(ch.alloc(), new byte[] { 1 }, 0, 1)).sync();
79 assertEquals(1, s.getInputStream().read());
80
81 assertTrue(h.ch.isOpen());
82 assertTrue(h.ch.isActive());
83 assertFalse(h.ch.isInputShutdown());
84 assertFalse(h.ch.isOutputShutdown());
85
86
87 ch.shutdownOutput().sync();
88 assertEquals(-1, s.getInputStream().read());
89
90 assertTrue(h.ch.isOpen());
91 assertTrue(h.ch.isActive());
92 assertFalse(h.ch.isInputShutdown());
93 assertTrue(h.ch.isOutputShutdown());
94
95
96 s.getOutputStream().write(new byte[] { 1 });
97 assertEquals(1, (int) h.queue.take());
98 } finally {
99 if (s != null) {
100 s.close();
101 }
102 if (ch != null) {
103 ch.close();
104 }
105 ss.close();
106 }
107 }
108
109 @Test
110 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
111 public void testShutdownOutputAfterClosed(TestInfo testInfo) throws Throwable {
112 run(testInfo, new Runner<Bootstrap>() {
113 @Override
114 public void run(Bootstrap bootstrap) throws Throwable {
115 testShutdownOutputAfterClosed(bootstrap);
116 }
117 });
118 }
119
120 public void testShutdownOutputAfterClosed(Bootstrap cb) throws Throwable {
121 TestHandler h = new TestHandler();
122 ServerSocket ss = new ServerSocket();
123 Socket s = null;
124 try {
125 ss.bind(newSocketAddress());
126 final SocketChannel ch = (SocketChannel) cb.handler(h).connect(ss.getLocalSocketAddress()).sync().channel();
127 assertTrue(ch.isActive());
128 s = ss.accept();
129
130 ch.close().syncUninterruptibly();
131 Throwable cause = assertThrows(Throwable.class, new Executable() {
132 @Override
133 public void execute() throws Throwable {
134 ch.shutdownInput().syncUninterruptibly();
135 }
136 });
137 checkThrowable(cause);
138 cause = assertThrows(Throwable.class, new Executable() {
139 @Override
140 public void execute() throws Throwable {
141 ch.shutdownOutput().syncUninterruptibly();
142 }
143 });
144 checkThrowable(cause);
145 } finally {
146 if (s != null) {
147 s.close();
148 }
149 ss.close();
150 }
151 }
152
153 @Disabled
154 @Test
155 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
156 public void testWriteAfterShutdownOutputNoWritabilityChange(TestInfo testInfo) throws Throwable {
157 run(testInfo, new Runner<Bootstrap>() {
158 @Override
159 public void run(Bootstrap bootstrap) throws Throwable {
160 testWriteAfterShutdownOutputNoWritabilityChange(bootstrap);
161 }
162 });
163 }
164
165 public void testWriteAfterShutdownOutputNoWritabilityChange(Bootstrap cb) throws Throwable {
166 final TestHandler h = new TestHandler();
167 ServerSocket ss = new ServerSocket();
168 Socket s = null;
169 SocketChannel ch = null;
170 try {
171 ss.bind(newSocketAddress());
172 cb.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(2, 4));
173 ch = (SocketChannel) cb.handler(h).connect(ss.getLocalSocketAddress()).sync().channel();
174 assumeFalse(ch instanceof OioSocketChannel);
175 assertTrue(ch.isActive());
176 assertFalse(ch.isOutputShutdown());
177
178 s = ss.accept();
179
180 byte[] expectedBytes = new byte[]{ 1, 2, 3, 4, 5, 6 };
181 ChannelFuture writeFuture = ch.write(Unpooled.wrappedBuffer(expectedBytes));
182 h.assertWritability(false);
183 ch.flush();
184 writeFuture.sync();
185 h.assertWritability(true);
186 for (int i = 0; i < expectedBytes.length; ++i) {
187 assertEquals(expectedBytes[i], s.getInputStream().read());
188 }
189
190 assertTrue(h.ch.isOpen());
191 assertTrue(h.ch.isActive());
192 assertFalse(h.ch.isInputShutdown());
193 assertFalse(h.ch.isOutputShutdown());
194
195
196 ch.shutdownOutput().sync();
197 assertEquals(-1, s.getInputStream().read());
198
199 assertTrue(h.ch.isOpen());
200 assertTrue(h.ch.isActive());
201 assertFalse(h.ch.isInputShutdown());
202 assertTrue(h.ch.isOutputShutdown());
203
204
205 final SocketChannel finalCh = ch;
206 Throwable cause = assertThrows(Throwable.class,
207 new Executable() {
208 @Override
209 public void execute() throws Throwable {
210 finalCh.writeAndFlush(randomBufferType(finalCh.alloc(), new byte[]{ 2 }, 0 , 2)).sync();
211 }
212 });
213 checkThrowable(cause);
214 assertNull(h.writabilityQueue.poll());
215 } finally {
216 if (s != null) {
217 s.close();
218 }
219 if (ch != null) {
220 ch.close();
221 }
222 ss.close();
223 }
224 }
225
226 @Test
227 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
228 public void testShutdownOutputSoLingerNoAssertError(TestInfo testInfo) throws Throwable {
229 run(testInfo, new Runner<Bootstrap>() {
230 @Override
231 public void run(Bootstrap bootstrap) throws Throwable {
232 testShutdownOutputSoLingerNoAssertError(bootstrap);
233 }
234 });
235 }
236
237 public void testShutdownOutputSoLingerNoAssertError(Bootstrap cb) throws Throwable {
238 testShutdownSoLingerNoAssertError0(cb, true);
239 }
240
241 @Test
242 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
243 public void testShutdownSoLingerNoAssertError(TestInfo testInfo) throws Throwable {
244 run(testInfo, new Runner<Bootstrap>() {
245 @Override
246 public void run(Bootstrap bootstrap) throws Throwable {
247 testShutdownSoLingerNoAssertError(bootstrap);
248 }
249 });
250 }
251
252 public void testShutdownSoLingerNoAssertError(Bootstrap cb) throws Throwable {
253 testShutdownSoLingerNoAssertError0(cb, false);
254 }
255
256 private void testShutdownSoLingerNoAssertError0(Bootstrap cb, boolean output) throws Throwable {
257 ServerSocket ss = new ServerSocket();
258 Socket s = null;
259
260 ChannelFuture cf = null;
261 try {
262 ss.bind(newSocketAddress());
263 cf = cb.option(ChannelOption.SO_LINGER, 1).handler(new ChannelInboundHandlerAdapter())
264 .connect(ss.getLocalSocketAddress()).sync();
265 s = ss.accept();
266
267 cf.sync();
268
269 if (output) {
270 ((SocketChannel) cf.channel()).shutdownOutput().sync();
271 } else {
272 ((SocketChannel) cf.channel()).shutdown().sync();
273 }
274 } finally {
275 if (s != null) {
276 s.close();
277 }
278 if (cf != null) {
279 cf.channel().close();
280 }
281 ss.close();
282 }
283 }
284 private static void checkThrowable(Throwable cause) throws Throwable {
285
286 if (!(cause instanceof ClosedChannelException) && !(cause instanceof SocketException)) {
287 throw cause;
288 }
289 }
290
291 private static final class TestHandler extends SimpleChannelInboundHandler<ByteBuf> {
292 volatile SocketChannel ch;
293 final BlockingQueue<Byte> queue = new LinkedBlockingQueue<Byte>();
294 final BlockingDeque<Boolean> writabilityQueue = new LinkedBlockingDeque<Boolean>();
295
296 @Override
297 public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
298 writabilityQueue.add(ctx.channel().isWritable());
299 }
300
301 @Override
302 public void channelActive(ChannelHandlerContext ctx) throws Exception {
303 ch = (SocketChannel) ctx.channel();
304 }
305
306 @Override
307 public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
308 queue.offer(msg.readByte());
309 }
310
311 private void drainWritabilityQueue() throws InterruptedException {
312 while ((writabilityQueue.poll(100, TimeUnit.MILLISECONDS)) != null) {
313
314 }
315 }
316
317 void assertWritability(boolean isWritable) throws InterruptedException {
318 try {
319 Boolean writability = writabilityQueue.takeLast();
320 assertEquals(isWritable, writability);
321
322 drainWritabilityQueue();
323 } catch (Throwable c) {
324 c.printStackTrace();
325 }
326 }
327 }
328 }