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