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 javax.net.ssl.SSLEngine;
33  import javax.net.ssl.SSLException;
34  
35  import org.eclipse.jetty.io.AsyncEndPoint;
36  import org.eclipse.jetty.io.nio.AsyncConnection;
37  import org.eclipse.jetty.io.nio.SslConnection;
38  import org.eclipse.jetty.npn.NextProtoNego;
39  import org.eclipse.jetty.server.nio.SelectChannelConnector;
40  import org.eclipse.jetty.spdy.api.SPDY;
41  import org.eclipse.jetty.spdy.api.Session;
42  import org.eclipse.jetty.spdy.api.server.ServerSessionFrameListener;
43  import org.eclipse.jetty.util.ssl.SslContextFactory;
44  import org.eclipse.jetty.util.thread.ThreadPool;
45  import org.slf4j.Logger;
46  import org.slf4j.LoggerFactory;
47  
48  public class SPDYServerConnector extends SelectChannelConnector
49  {
50      private static final Logger logger = LoggerFactory.getLogger(SPDYServerConnector.class);
51  
52      // Order is important on server side, so we use a LinkedHashMap
53      private final Map<String, AsyncConnectionFactory> factories = new LinkedHashMap<>();
54      private final Queue<Session> sessions = new ConcurrentLinkedQueue<>();
55      private final ByteBufferPool bufferPool = new StandardByteBufferPool();
56      private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
57      private final ServerSessionFrameListener listener;
58      private final SslContextFactory sslContextFactory;
59      private AsyncConnectionFactory defaultConnectionFactory;
60  
61      public SPDYServerConnector(ServerSessionFrameListener listener)
62      {
63          this(listener, null);
64      }
65  
66      public SPDYServerConnector(ServerSessionFrameListener listener, SslContextFactory sslContextFactory)
67      {
68          this.listener = listener;
69          this.sslContextFactory = sslContextFactory;
70          if (sslContextFactory != null)
71              addBean(sslContextFactory);
72      }
73  
74      public ByteBufferPool getByteBufferPool()
75      {
76          return bufferPool;
77      }
78  
79      protected Executor getExecutor()
80      {
81          final ThreadPool threadPool = getThreadPool();
82          if (threadPool instanceof Executor)
83              return (Executor)threadPool;
84          return new Executor()
85          {
86              @Override
87              public void execute(Runnable command)
88              {
89                  threadPool.dispatch(command);
90              }
91          };
92      }
93  
94      protected ScheduledExecutorService getScheduler()
95      {
96          return scheduler;
97      }
98  
99      @Override
100     protected void doStart() throws Exception
101     {
102         super.doStart();
103         defaultConnectionFactory = new ServerSPDYAsyncConnectionFactory(SPDY.V2, getByteBufferPool(), getExecutor(), scheduler, listener);
104         putAsyncConnectionFactory("spdy/2", defaultConnectionFactory);
105         logger.info("SPDY support is experimental. Please report feedback at jetty-dev@eclipse.org");
106     }
107 
108     @Override
109     protected void doStop() throws Exception
110     {
111         closeSessions();
112         scheduler.shutdown();
113         super.doStop();
114     }
115 
116     @Override
117     public void join() throws InterruptedException
118     {
119         scheduler.awaitTermination(0, TimeUnit.MILLISECONDS);
120         super.join();
121     }
122 
123     public AsyncConnectionFactory getAsyncConnectionFactory(String protocol)
124     {
125         synchronized (factories)
126         {
127             return factories.get(protocol);
128         }
129     }
130 
131     public AsyncConnectionFactory putAsyncConnectionFactory(String protocol, AsyncConnectionFactory factory)
132     {
133         synchronized (factories)
134         {
135             return factories.put(protocol, factory);
136         }
137     }
138 
139     public AsyncConnectionFactory removeAsyncConnectionFactory(String protocol)
140     {
141         synchronized (factories)
142         {
143             return factories.remove(protocol);
144         }
145     }
146 
147     public Map<String, AsyncConnectionFactory> getAsyncConnectionFactories()
148     {
149         synchronized (factories)
150         {
151             return new LinkedHashMap<>(factories);
152         }
153     }
154 
155     protected List<String> provideProtocols()
156     {
157         synchronized (factories)
158         {
159             return new ArrayList<>(factories.keySet());
160         }
161     }
162 
163     protected AsyncConnectionFactory getDefaultAsyncConnectionFactory()
164     {
165         return defaultConnectionFactory;
166     }
167 
168     @Override
169     protected AsyncConnection newConnection(final SocketChannel channel, AsyncEndPoint endPoint)
170     {
171         if (sslContextFactory != null)
172         {
173             SSLEngine engine = newSSLEngine(sslContextFactory, channel);
174             SslConnection sslConnection = new SslConnection(engine, endPoint);
175             endPoint.setConnection(sslConnection);
176             final AsyncEndPoint sslEndPoint = sslConnection.getSslEndPoint();
177 
178             NextProtoNego.put(engine, new NextProtoNego.ServerProvider()
179             {
180                 @Override
181                 public void unsupported()
182                 {
183                     AsyncConnectionFactory connectionFactory = getDefaultAsyncConnectionFactory();
184                     AsyncConnection connection = connectionFactory.newAsyncConnection(channel, sslEndPoint, SPDYServerConnector.this);
185                     sslEndPoint.setConnection(connection);
186                 }
187 
188                 @Override
189                 public List<String> protocols()
190                 {
191                     return provideProtocols();
192                 }
193 
194                 @Override
195                 public void protocolSelected(String protocol)
196                 {
197                     AsyncConnectionFactory connectionFactory = getAsyncConnectionFactory(protocol);
198                     AsyncConnection connection = connectionFactory.newAsyncConnection(channel, sslEndPoint, SPDYServerConnector.this);
199                     sslEndPoint.setConnection(connection);
200                 }
201             });
202 
203             AsyncConnection connection = new EmptyAsyncConnection(sslEndPoint);
204             sslEndPoint.setConnection(connection);
205 
206             startHandshake(engine);
207 
208             return sslConnection;
209         }
210         else
211         {
212             AsyncConnectionFactory connectionFactory = getDefaultAsyncConnectionFactory();
213             AsyncConnection connection = connectionFactory.newAsyncConnection(channel, endPoint, this);
214             endPoint.setConnection(connection);
215             return connection;
216         }
217     }
218 
219     protected SSLEngine newSSLEngine(SslContextFactory sslContextFactory, SocketChannel channel)
220     {
221         String peerHost = channel.socket().getInetAddress().getHostAddress();
222         int peerPort = channel.socket().getPort();
223         SSLEngine engine = sslContextFactory.newSslEngine(peerHost, peerPort);
224         engine.setUseClientMode(false);
225         return engine;
226     }
227 
228     private void startHandshake(SSLEngine engine)
229     {
230         try
231         {
232             engine.beginHandshake();
233         }
234         catch (SSLException x)
235         {
236             throw new RuntimeException(x);
237         }
238     }
239 
240     protected boolean sessionOpened(Session session)
241     {
242         // Add sessions only if the connector is not stopping
243         return isRunning() && sessions.offer(session);
244     }
245 
246     protected boolean sessionClosed(Session session)
247     {
248         // Remove sessions only if the connector is not stopping
249         // to avoid concurrent removes during iterations
250         return isRunning() && sessions.remove(session);
251     }
252 
253     private void closeSessions()
254     {
255         for (Session session : sessions)
256             session.goAway();
257         sessions.clear();
258     }
259 
260     protected Collection<Session> getSessions()
261     {
262         return Collections.unmodifiableCollection(sessions);
263     }
264 }