View Javadoc
1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  
28  package org.apache.http.impl.nio.reactor;
29  
30  import java.io.IOException;
31  import java.net.BindException;
32  import java.net.InetSocketAddress;
33  import java.util.Set;
34  import java.util.concurrent.CountDownLatch;
35  import java.util.concurrent.TimeUnit;
36  
37  import org.apache.http.HttpResponseInterceptor;
38  import org.apache.http.config.ConnectionConfig;
39  import org.apache.http.impl.nio.DefaultHttpServerIODispatch;
40  import org.apache.http.nio.protocol.HttpAsyncService;
41  import org.apache.http.nio.protocol.UriHttpAsyncRequestHandlerMapper;
42  import org.apache.http.nio.reactor.IOEventDispatch;
43  import org.apache.http.nio.reactor.IOReactorExceptionHandler;
44  import org.apache.http.nio.reactor.IOReactorStatus;
45  import org.apache.http.nio.reactor.ListenerEndpoint;
46  import org.apache.http.nio.reactor.ListeningIOReactor;
47  import org.apache.http.protocol.HttpProcessor;
48  import org.apache.http.protocol.ImmutableHttpProcessor;
49  import org.apache.http.protocol.ResponseConnControl;
50  import org.apache.http.protocol.ResponseContent;
51  import org.apache.http.protocol.ResponseDate;
52  import org.apache.http.protocol.ResponseServer;
53  import org.junit.Assert;
54  import org.junit.Test;
55  
56  /**
57   * Basic tests for {@link DefaultListeningIOReactor}.
58   */
59  public class TestDefaultListeningIOReactor {
60  
61      private static IOEventDispatch createIOEventDispatch() {
62          final HttpProcessor httpproc = new ImmutableHttpProcessor(new HttpResponseInterceptor[] {
63                  new ResponseDate(),
64                  new ResponseServer(),
65                  new ResponseContent(),
66                  new ResponseConnControl()
67          });
68          final HttpAsyncService serviceHandler = new HttpAsyncService(httpproc,
69                  new UriHttpAsyncRequestHandlerMapper());
70          return new DefaultHttpServerIODispatch(serviceHandler, ConnectionConfig.DEFAULT);
71      }
72  
73      @Test
74      public void testEndpointUpAndDown() throws Exception {
75          final IOEventDispatch eventDispatch = createIOEventDispatch();
76          final IOReactorConfig config = IOReactorConfig.custom().setIoThreadCount(1).build();
77          final ListeningIOReactor ioreactor = new DefaultListeningIOReactor(config);
78  
79          final Thread t = new Thread(new Runnable() {
80  
81              @Override
82              public void run() {
83                  try {
84                      ioreactor.execute(eventDispatch);
85                  } catch (final IOException ex) {
86                  }
87              }
88  
89          });
90  
91          t.start();
92  
93          Set<ListenerEndpoint> endpoints = ioreactor.getEndpoints();
94          Assert.assertNotNull(endpoints);
95          Assert.assertEquals(0, endpoints.size());
96  
97          final ListenerEndpoint endpoint1 = ioreactor.listen(new InetSocketAddress(0));
98          endpoint1.waitFor();
99  
100         final ListenerEndpoint endpoint2 = ioreactor.listen(new InetSocketAddress(0));
101         endpoint2.waitFor();
102         final int port = ((InetSocketAddress) endpoint2.getAddress()).getPort();
103 
104         endpoints = ioreactor.getEndpoints();
105         Assert.assertNotNull(endpoints);
106         Assert.assertEquals(2, endpoints.size());
107 
108         endpoint1.close();
109 
110         endpoints = ioreactor.getEndpoints();
111         Assert.assertNotNull(endpoints);
112         Assert.assertEquals(1, endpoints.size());
113 
114         final ListenerEndpoint endpoint = endpoints.iterator().next();
115 
116         Assert.assertEquals(port, ((InetSocketAddress) endpoint.getAddress()).getPort());
117 
118         ioreactor.shutdown(1000);
119         t.join(1000);
120 
121         Assert.assertEquals(IOReactorStatus.SHUT_DOWN, ioreactor.getStatus());
122     }
123 
124     @Test
125     public void testEndpointAlreadyBoundFatal() throws Exception {
126         final IOEventDispatch eventDispatch = createIOEventDispatch();
127         final IOReactorConfig config = IOReactorConfig.custom().setIoThreadCount(1).build();
128         final ListeningIOReactor ioreactor = new DefaultListeningIOReactor(config);
129 
130         final CountDownLatch latch = new CountDownLatch(1);
131 
132         final Thread t = new Thread(new Runnable() {
133 
134             @Override
135             public void run() {
136                 try {
137                     ioreactor.execute(eventDispatch);
138                     Assert.fail("IOException should have been thrown");
139                 } catch (final IOException ex) {
140                     latch.countDown();
141                 }
142             }
143 
144         });
145 
146         t.start();
147 
148         final ListenerEndpoint endpoint1 = ioreactor.listen(new InetSocketAddress(0));
149         endpoint1.waitFor();
150         final int port = ((InetSocketAddress) endpoint1.getAddress()).getPort();
151 
152         final ListenerEndpoint endpoint2 = ioreactor.listen(new InetSocketAddress(port));
153         endpoint2.waitFor();
154         Assert.assertNotNull(endpoint2.getException());
155 
156         // I/O reactor is now expected to be shutting down
157         latch.await(2000, TimeUnit.MILLISECONDS);
158         Assert.assertTrue(ioreactor.getStatus().compareTo(IOReactorStatus.SHUTTING_DOWN) >= 0);
159 
160         final Set<ListenerEndpoint> endpoints = ioreactor.getEndpoints();
161         Assert.assertNotNull(endpoints);
162         Assert.assertEquals(0, endpoints.size());
163 
164         ioreactor.shutdown(1000);
165         t.join(1000);
166 
167         Assert.assertEquals(IOReactorStatus.SHUT_DOWN, ioreactor.getStatus());
168     }
169 
170     @Test
171     public void testEndpointAlreadyBoundNonFatal() throws Exception {
172         final IOEventDispatch eventDispatch = createIOEventDispatch();
173         final IOReactorConfig config = IOReactorConfig.custom().setIoThreadCount(1).build();
174         final DefaultListeningIOReactor ioreactor = new DefaultListeningIOReactor(config);
175 
176         ioreactor.setExceptionHandler(new IOReactorExceptionHandler() {
177 
178             @Override
179             public boolean handle(final IOException ex) {
180                 return (ex instanceof BindException);
181             }
182 
183             @Override
184             public boolean handle(final RuntimeException ex) {
185                 return false;
186             }
187 
188         });
189 
190         final Thread t = new Thread(new Runnable() {
191 
192             @Override
193             public void run() {
194                 try {
195                     ioreactor.execute(eventDispatch);
196                 } catch (final IOException ex) {
197                 }
198             }
199 
200         });
201 
202         t.start();
203 
204         final ListenerEndpoint endpoint1 = ioreactor.listen(new InetSocketAddress(9999));
205         endpoint1.waitFor();
206 
207         final ListenerEndpoint endpoint2 = ioreactor.listen(new InetSocketAddress(9999));
208         endpoint2.waitFor();
209         Assert.assertNotNull(endpoint2.getException());
210 
211         // Sleep a little to make sure the I/O reactor is not shutting down
212         Thread.sleep(500);
213 
214         Assert.assertEquals(IOReactorStatus.ACTIVE, ioreactor.getStatus());
215 
216         ioreactor.shutdown(1000);
217         t.join(1000);
218 
219         Assert.assertEquals(IOReactorStatus.SHUT_DOWN, ioreactor.getStatus());
220     }
221 
222 }