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