1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.spdy.client;
20
21 import java.io.IOException;
22 import java.net.SocketAddress;
23 import java.nio.channels.SelectionKey;
24 import java.nio.channels.SocketChannel;
25 import java.util.Collection;
26 import java.util.Collections;
27 import java.util.List;
28 import java.util.Queue;
29 import java.util.concurrent.ConcurrentLinkedQueue;
30 import java.util.concurrent.ExecutionException;
31 import java.util.concurrent.Executor;
32 import javax.net.ssl.SSLEngine;
33
34 import org.eclipse.jetty.io.ByteBufferPool;
35 import org.eclipse.jetty.io.Connection;
36 import org.eclipse.jetty.io.EndPoint;
37 import org.eclipse.jetty.io.MappedByteBufferPool;
38 import org.eclipse.jetty.io.SelectChannelEndPoint;
39 import org.eclipse.jetty.io.SelectorManager;
40 import org.eclipse.jetty.io.ssl.SslConnection;
41 import org.eclipse.jetty.io.ssl.SslConnection.DecryptedEndPoint;
42 import org.eclipse.jetty.spdy.FlowControlStrategy;
43 import org.eclipse.jetty.spdy.api.GoAwayInfo;
44 import org.eclipse.jetty.spdy.api.Session;
45 import org.eclipse.jetty.spdy.api.SessionFrameListener;
46 import org.eclipse.jetty.util.Callback;
47 import org.eclipse.jetty.util.FuturePromise;
48 import org.eclipse.jetty.util.Promise;
49 import org.eclipse.jetty.util.component.ContainerLifeCycle;
50 import org.eclipse.jetty.util.ssl.SslContextFactory;
51 import org.eclipse.jetty.util.thread.QueuedThreadPool;
52 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
53 import org.eclipse.jetty.util.thread.Scheduler;
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71 public class SPDYClient
72 {
73 private final SPDYClientConnectionFactory connectionFactory = new SPDYClientConnectionFactory();
74 final short version;
75 final Factory factory;
76 private volatile SocketAddress bindAddress;
77 private volatile long idleTimeout = -1;
78 private volatile int initialWindowSize;
79 private volatile boolean executeOnFillable;
80
81 protected SPDYClient(short version, Factory factory)
82 {
83 this.version = version;
84 this.factory = factory;
85 setInitialWindowSize(65536);
86 }
87
88
89
90
91
92 public SocketAddress getBindAddress()
93 {
94 return bindAddress;
95 }
96
97
98
99
100
101 public void setBindAddress(SocketAddress bindAddress)
102 {
103 this.bindAddress = bindAddress;
104 }
105
106
107
108
109
110
111
112
113
114
115
116
117 public Session connect(SocketAddress address, SessionFrameListener listener) throws ExecutionException, InterruptedException
118 {
119 FuturePromise<Session> promise = new FuturePromise<>();
120 connect(address, listener, promise);
121 return promise.get();
122 }
123
124
125
126
127
128
129
130
131
132
133
134
135 public void connect(SocketAddress address, SessionFrameListener listener, Promise<Session> promise)
136 {
137 if (!factory.isStarted())
138 throw new IllegalStateException(Factory.class.getSimpleName() + " is not started");
139
140 try
141 {
142 SocketChannel channel = SocketChannel.open();
143 if (bindAddress != null)
144 channel.bind(bindAddress);
145 channel.socket().setTcpNoDelay(true);
146 channel.configureBlocking(false);
147
148 SessionPromise result = new SessionPromise(promise, channel, this, listener);
149
150 channel.connect(address);
151 factory.selector.connect(channel, result);
152 }
153 catch (IOException x)
154 {
155 promise.failed(x);
156 }
157 }
158
159 public long getIdleTimeout()
160 {
161 return idleTimeout;
162 }
163
164 public void setIdleTimeout(long idleTimeout)
165 {
166 this.idleTimeout = idleTimeout;
167 }
168
169 public int getInitialWindowSize()
170 {
171 return initialWindowSize;
172 }
173
174 public void setInitialWindowSize(int initialWindowSize)
175 {
176 this.initialWindowSize = initialWindowSize;
177 }
178
179 public boolean isExecuteOnFillable()
180 {
181 return executeOnFillable;
182 }
183
184 public void setExecuteOnFillable(boolean executeOnFillable)
185 {
186 this.executeOnFillable = executeOnFillable;
187 }
188
189 protected String selectProtocol(List<String> serverProtocols)
190 {
191 String protocol = "spdy/" + version;
192 for (String serverProtocol : serverProtocols)
193 {
194 if (serverProtocol.equals(protocol))
195 return protocol;
196 }
197 return null;
198 }
199
200 public SPDYClientConnectionFactory getConnectionFactory()
201 {
202 return connectionFactory;
203 }
204
205 protected SSLEngine newSSLEngine(SslContextFactory sslContextFactory, SocketChannel channel)
206 {
207 String peerHost = channel.socket().getInetAddress().getHostName();
208 int peerPort = channel.socket().getPort();
209 SSLEngine engine = sslContextFactory.newSSLEngine(peerHost, peerPort);
210 engine.setUseClientMode(true);
211 return engine;
212 }
213
214 protected FlowControlStrategy newFlowControlStrategy()
215 {
216 return FlowControlStrategyFactory.newFlowControlStrategy(version);
217 }
218
219 public static class Factory extends ContainerLifeCycle
220 {
221 private final Queue<Session> sessions = new ConcurrentLinkedQueue<>();
222 private final ByteBufferPool bufferPool = new MappedByteBufferPool();
223 private final Scheduler scheduler;
224 private final Executor executor;
225 private final SslContextFactory sslContextFactory;
226 private final SelectorManager selector;
227 private final long idleTimeout;
228 private long connectTimeout = 15000;
229
230 public Factory()
231 {
232 this(null, null);
233 }
234
235 public Factory(SslContextFactory sslContextFactory)
236 {
237 this(null, null, sslContextFactory);
238 }
239
240 public Factory(Executor executor)
241 {
242 this(executor, null);
243 }
244
245 public Factory(Executor executor, Scheduler scheduler)
246 {
247 this(executor, scheduler, null);
248 }
249
250 public Factory(Executor executor, Scheduler scheduler, SslContextFactory sslContextFactory)
251 {
252 this(executor, scheduler, sslContextFactory, 30000);
253 }
254
255 public Factory(Executor executor, Scheduler scheduler, SslContextFactory sslContextFactory, long idleTimeout)
256 {
257 this.idleTimeout = idleTimeout;
258
259 if (executor == null)
260 executor = new QueuedThreadPool();
261 this.executor = executor;
262 addBean(executor);
263
264 if (scheduler == null)
265 scheduler = new ScheduledExecutorScheduler();
266 this.scheduler = scheduler;
267 addBean(scheduler);
268
269 this.sslContextFactory = sslContextFactory;
270 if (sslContextFactory != null)
271 addBean(sslContextFactory);
272
273 selector = new ClientSelectorManager(executor, scheduler);
274 selector.setConnectTimeout(getConnectTimeout());
275 addBean(selector);
276 }
277
278 public ByteBufferPool getByteBufferPool()
279 {
280 return bufferPool;
281 }
282
283 public Scheduler getScheduler()
284 {
285 return scheduler;
286 }
287
288 public Executor getExecutor()
289 {
290 return executor;
291 }
292
293 public long getConnectTimeout()
294 {
295 return connectTimeout;
296 }
297
298 public void setConnectTimeout(long connectTimeout)
299 {
300 this.connectTimeout = connectTimeout;
301 }
302
303 public SPDYClient newSPDYClient(short version)
304 {
305 return new SPDYClient(version, this);
306 }
307
308 @Override
309 protected void doStop() throws Exception
310 {
311 closeConnections();
312 super.doStop();
313 }
314
315 boolean sessionOpened(Session session)
316 {
317
318 return isRunning() && sessions.offer(session);
319 }
320
321 boolean sessionClosed(Session session)
322 {
323
324
325 return isRunning() && sessions.remove(session);
326 }
327
328 private void closeConnections()
329 {
330 for (Session session : sessions)
331 session.goAway(new GoAwayInfo(), new Callback.Adapter());
332 sessions.clear();
333 }
334
335 public Collection<Session> getSessions()
336 {
337 return Collections.unmodifiableCollection(sessions);
338 }
339
340 private class ClientSelectorManager extends SelectorManager
341 {
342 private ClientSelectorManager(Executor executor, Scheduler scheduler)
343 {
344 super(executor, scheduler);
345 }
346
347 @Override
348 protected EndPoint newEndPoint(SocketChannel channel, ManagedSelector selectSet, SelectionKey key) throws IOException
349 {
350 SessionPromise attachment = (SessionPromise)key.attachment();
351
352 long clientIdleTimeout = attachment.client.getIdleTimeout();
353 if (clientIdleTimeout < 0)
354 clientIdleTimeout = idleTimeout;
355
356 return new SelectChannelEndPoint(channel, selectSet, key, getScheduler(), clientIdleTimeout);
357 }
358
359 @Override
360 public Connection newConnection(final SocketChannel channel, EndPoint endPoint, final Object attachment)
361 {
362 SessionPromise sessionPromise = (SessionPromise)attachment;
363 final SPDYClient client = sessionPromise.client;
364
365 try
366 {
367 if (sslContextFactory != null)
368 {
369 final SSLEngine engine = client.newSSLEngine(sslContextFactory, channel);
370 SslConnection sslConnection = new SslConnection(bufferPool, getExecutor(), endPoint, engine);
371 sslConnection.setRenegotiationAllowed(sslContextFactory.isRenegotiationAllowed());
372 DecryptedEndPoint sslEndPoint = sslConnection.getDecryptedEndPoint();
373 NextProtoNegoClientConnection connection = new NextProtoNegoClientConnection(channel, sslEndPoint, attachment, getExecutor(), client);
374 sslEndPoint.setConnection(connection);
375 return sslConnection;
376 }
377
378 SPDYClientConnectionFactory connectionFactory = new SPDYClientConnectionFactory();
379 return connectionFactory.newConnection(channel, endPoint, attachment);
380 }
381 catch (RuntimeException x)
382 {
383 sessionPromise.failed(x);
384 throw x;
385 }
386 }
387 }
388 }
389
390 static class SessionPromise extends FuturePromise<Session>
391 {
392 private final SocketChannel channel;
393 private final Promise<Session> wrappedPromise;
394 final SPDYClient client;
395 final SessionFrameListener listener;
396
397 private SessionPromise(Promise<Session> promise, SocketChannel channel, SPDYClient client,
398 SessionFrameListener listener)
399 {
400 this.channel = channel;
401 this.client = client;
402 this.listener = listener;
403 this.wrappedPromise = promise;
404 }
405
406 @Override
407 public void succeeded(Session result)
408 {
409 wrappedPromise.succeeded(result);
410 super.succeeded(result);
411 }
412
413 @Override
414 public void failed(Throwable cause)
415 {
416 wrappedPromise.failed(cause);
417 super.failed(cause);
418 }
419
420 @Override
421 public boolean cancel(boolean mayInterruptIfRunning)
422 {
423 try
424 {
425 super.cancel(mayInterruptIfRunning);
426 channel.close();
427 return true;
428 }
429 catch (IOException x)
430 {
431 return true;
432 }
433 }
434 }
435 }