View Javadoc

1   /*
2    * Copyright (c) 2012 the original author or authors.
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    *
8    *     http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  
17  package org.eclipse.jetty.spdy;
18  
19  import java.nio.channels.SocketChannel;
20  import java.util.ArrayList;
21  import java.util.Collection;
22  import java.util.Collections;
23  import java.util.LinkedHashMap;
24  import java.util.List;
25  import java.util.Map;
26  import java.util.Queue;
27  import java.util.concurrent.ConcurrentLinkedQueue;
28  import java.util.concurrent.Executor;
29  import java.util.concurrent.Executors;
30  import java.util.concurrent.ScheduledExecutorService;
31  import java.util.concurrent.TimeUnit;
32  import java.util.concurrent.atomic.AtomicReference;
33  import javax.net.ssl.SSLEngine;
34  import javax.net.ssl.SSLException;
35  
36  import org.eclipse.jetty.io.AsyncEndPoint;
37  import org.eclipse.jetty.io.nio.AsyncConnection;
38  import org.eclipse.jetty.io.nio.SslConnection;
39  import org.eclipse.jetty.npn.NextProtoNego;
40  import org.eclipse.jetty.server.nio.SelectChannelConnector;
41  import org.eclipse.jetty.spdy.api.SPDY;
42  import org.eclipse.jetty.spdy.api.Session;
43  import org.eclipse.jetty.spdy.api.server.ServerSessionFrameListener;
44  import org.eclipse.jetty.util.log.Log;
45  import org.eclipse.jetty.util.log.Logger;
46  import org.eclipse.jetty.util.ssl.SslContextFactory;
47  import org.eclipse.jetty.util.thread.ThreadPool;
48  
49  public class SPDYServerConnector extends SelectChannelConnector
50  {
51      private static final Logger logger = Log.getLogger(SPDYServerConnector.class);
52  
53      // Order is important on server side, so we use a LinkedHashMap
54      private final Map<String, AsyncConnectionFactory> factories = new LinkedHashMap<>();
55      private final Queue<Session> sessions = new ConcurrentLinkedQueue<>();
56      private final ByteBufferPool bufferPool = new StandardByteBufferPool();
57      private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
58      private final ServerSessionFrameListener listener;
59      private final SslContextFactory sslContextFactory;
60      private AsyncConnectionFactory defaultConnectionFactory;
61  
62      public SPDYServerConnector(ServerSessionFrameListener listener)
63      {
64          this(listener, null);
65      }
66  
67      public SPDYServerConnector(ServerSessionFrameListener listener, SslContextFactory sslContextFactory)
68      {
69          this.listener = listener;
70          this.sslContextFactory = sslContextFactory;
71          if (sslContextFactory != null)
72              addBean(sslContextFactory);
73      }
74  
75      public ByteBufferPool getByteBufferPool()
76      {
77          return bufferPool;
78      }
79  
80      public Executor getExecutor()
81      {
82          final ThreadPool threadPool = getThreadPool();
83          if (threadPool instanceof Executor)
84              return (Executor)threadPool;
85          return new Executor()
86          {
87              @Override
88              public void execute(Runnable command)
89              {
90                  threadPool.dispatch(command);
91              }
92          };
93      }
94  
95      public ScheduledExecutorService getScheduler()
96      {
97          return scheduler;
98      }
99  
100     public SslContextFactory getSslContextFactory()
101     {
102         return sslContextFactory;
103     }
104 
105     @Override
106     protected void doStart() throws Exception
107     {
108         super.doStart();
109         defaultConnectionFactory = new ServerSPDYAsyncConnectionFactory(SPDY.V2, getByteBufferPool(), getExecutor(), scheduler, listener);
110         putAsyncConnectionFactory("spdy/2", defaultConnectionFactory);
111         logger.info("SPDY support is experimental. Please report feedback at jetty-dev@eclipse.org");
112     }
113 
114     @Override
115     protected void doStop() throws Exception
116     {
117         closeSessions();
118         scheduler.shutdown();
119         super.doStop();
120     }
121 
122     @Override
123     public void join() throws InterruptedException
124     {
125         scheduler.awaitTermination(0, TimeUnit.MILLISECONDS);
126         super.join();
127     }
128 
129     public AsyncConnectionFactory getAsyncConnectionFactory(String protocol)
130     {
131         synchronized (factories)
132         {
133             return factories.get(protocol);
134         }
135     }
136 
137     public AsyncConnectionFactory putAsyncConnectionFactory(String protocol, AsyncConnectionFactory factory)
138     {
139         synchronized (factories)
140         {
141             return factories.put(protocol, factory);
142         }
143     }
144 
145     public AsyncConnectionFactory removeAsyncConnectionFactory(String protocol)
146     {
147         synchronized (factories)
148         {
149             return factories.remove(protocol);
150         }
151     }
152 
153     public Map<String, AsyncConnectionFactory> getAsyncConnectionFactories()
154     {
155         synchronized (factories)
156         {
157             return new LinkedHashMap<>(factories);
158         }
159     }
160 
161     protected List<String> provideProtocols()
162     {
163         synchronized (factories)
164         {
165             return new ArrayList<>(factories.keySet());
166         }
167     }
168 
169     protected AsyncConnectionFactory getDefaultAsyncConnectionFactory()
170     {
171         return defaultConnectionFactory;
172     }
173 
174     @Override
175     protected AsyncConnection newConnection(final SocketChannel channel, AsyncEndPoint endPoint)
176     {
177         if (sslContextFactory != null)
178         {
179             SSLEngine engine = newSSLEngine(sslContextFactory, channel);
180             final AtomicReference<AsyncEndPoint> sslEndPointRef = new AtomicReference<>();
181             SslConnection sslConnection = new SslConnection(engine, endPoint)
182             {
183                 @Override
184                 public void onClose()
185                 {
186                     sslEndPointRef.set(null);
187                     super.onClose();
188                 }
189             };
190             endPoint.setConnection(sslConnection);
191             AsyncEndPoint sslEndPoint = sslConnection.getSslEndPoint();
192             sslEndPointRef.set(sslEndPoint);
193 
194             // Instances of the ServerProvider inner class strong reference the
195             // SslEndPoint (via lexical scoping), which strong references the SSLEngine.
196             // Since NextProtoNego stores in a WeakHashMap the SSLEngine as key
197             // and this instance as value, we are in the situation where the value
198             // of a WeakHashMap refers indirectly to the key, which is bad because
199             // the entry will never be removed from the WeakHashMap.
200             // We use AtomicReferences to be captured via lexical scoping,
201             // and we null them out above when the connection is closed.
202             NextProtoNego.put(engine, new NextProtoNego.ServerProvider()
203             {
204                 @Override
205                 public void unsupported()
206                 {
207                     AsyncConnectionFactory connectionFactory = getDefaultAsyncConnectionFactory();
208                     AsyncEndPoint sslEndPoint = sslEndPointRef.get();
209                     AsyncConnection connection = connectionFactory.newAsyncConnection(channel, sslEndPoint, SPDYServerConnector.this);
210                     sslEndPoint.setConnection(connection);
211                 }
212 
213                 @Override
214                 public List<String> protocols()
215                 {
216                     return provideProtocols();
217                 }
218 
219                 @Override
220                 public void protocolSelected(String protocol)
221                 {
222                     AsyncConnectionFactory connectionFactory = getAsyncConnectionFactory(protocol);
223                     AsyncEndPoint sslEndPoint = sslEndPointRef.get();
224                     AsyncConnection connection = connectionFactory.newAsyncConnection(channel, sslEndPoint, SPDYServerConnector.this);
225                     sslEndPoint.setConnection(connection);
226                 }
227             });
228 
229             AsyncConnection connection = new EmptyAsyncConnection(sslEndPoint);
230             sslEndPoint.setConnection(connection);
231 
232             startHandshake(engine);
233 
234             return sslConnection;
235         }
236         else
237         {
238             AsyncConnectionFactory connectionFactory = getDefaultAsyncConnectionFactory();
239             AsyncConnection connection = connectionFactory.newAsyncConnection(channel, endPoint, this);
240             endPoint.setConnection(connection);
241             return connection;
242         }
243     }
244 
245     protected SSLEngine newSSLEngine(SslContextFactory sslContextFactory, SocketChannel channel)
246     {
247         String peerHost = channel.socket().getInetAddress().getHostAddress();
248         int peerPort = channel.socket().getPort();
249         SSLEngine engine = sslContextFactory.newSslEngine(peerHost, peerPort);
250         engine.setUseClientMode(false);
251         return engine;
252     }
253 
254     private void startHandshake(SSLEngine engine)
255     {
256         try
257         {
258             engine.beginHandshake();
259         }
260         catch (SSLException x)
261         {
262             throw new RuntimeException(x);
263         }
264     }
265 
266     protected boolean sessionOpened(Session session)
267     {
268         // Add sessions only if the connector is not stopping
269         return isRunning() && sessions.offer(session);
270     }
271 
272     protected boolean sessionClosed(Session session)
273     {
274         // Remove sessions only if the connector is not stopping
275         // to avoid concurrent removes during iterations
276         return isRunning() && sessions.remove(session);
277     }
278 
279     private void closeSessions()
280     {
281         for (Session session : sessions)
282             session.goAway();
283         sessions.clear();
284     }
285 
286     protected Collection<Session> getSessions()
287     {
288         return Collections.unmodifiableCollection(sessions);
289     }
290 }