1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.http2.client;
20
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.nio.channels.SelectableChannel;
24 import java.nio.channels.SelectionKey;
25 import java.nio.channels.SocketChannel;
26 import java.util.Arrays;
27 import java.util.HashMap;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.concurrent.Executor;
31
32 import org.eclipse.jetty.alpn.client.ALPNClientConnectionFactory;
33 import org.eclipse.jetty.http2.BufferingFlowControlStrategy;
34 import org.eclipse.jetty.http2.FlowControlStrategy;
35 import org.eclipse.jetty.http2.api.Session;
36 import org.eclipse.jetty.io.ByteBufferPool;
37 import org.eclipse.jetty.io.ClientConnectionFactory;
38 import org.eclipse.jetty.io.Connection;
39 import org.eclipse.jetty.io.EndPoint;
40 import org.eclipse.jetty.io.ManagedSelector;
41 import org.eclipse.jetty.io.MappedByteBufferPool;
42 import org.eclipse.jetty.io.SelectorManager;
43 import org.eclipse.jetty.io.SocketChannelEndPoint;
44 import org.eclipse.jetty.io.ssl.SslClientConnectionFactory;
45 import org.eclipse.jetty.util.Promise;
46 import org.eclipse.jetty.util.annotation.ManagedAttribute;
47 import org.eclipse.jetty.util.annotation.ManagedObject;
48 import org.eclipse.jetty.util.component.ContainerLifeCycle;
49 import org.eclipse.jetty.util.ssl.SslContextFactory;
50 import org.eclipse.jetty.util.thread.ExecutionStrategy;
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 import org.eclipse.jetty.util.thread.strategy.ProduceConsume;
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116 @ManagedObject
117 public class HTTP2Client extends ContainerLifeCycle
118 {
119 private Executor executor;
120 private Scheduler scheduler;
121 private ByteBufferPool bufferPool;
122 private ClientConnectionFactory connectionFactory;
123 private SelectorManager selector;
124 private int selectors = 1;
125 private long idleTimeout = 30000;
126 private long connectTimeout = 10000;
127 private int inputBufferSize = 8192;
128 private List<String> protocols = Arrays.asList("h2", "h2-17", "h2-16", "h2-15", "h2-14");
129 private int initialSessionRecvWindow = FlowControlStrategy.DEFAULT_WINDOW_SIZE;
130 private int initialStreamRecvWindow = FlowControlStrategy.DEFAULT_WINDOW_SIZE;
131 private FlowControlStrategy.Factory flowControlStrategyFactory = () -> new BufferingFlowControlStrategy(0.5F);
132 private ExecutionStrategy.Factory executionStrategyFactory = new ProduceConsume.Factory();
133
134 @Override
135 protected void doStart() throws Exception
136 {
137 if (executor == null)
138 setExecutor(new QueuedThreadPool());
139
140 if (scheduler == null)
141 setScheduler(new ScheduledExecutorScheduler());
142
143 if (bufferPool == null)
144 setByteBufferPool(new MappedByteBufferPool());
145
146 if (connectionFactory == null)
147 {
148 HTTP2ClientConnectionFactory h2 = new HTTP2ClientConnectionFactory();
149 setClientConnectionFactory((endPoint, context) ->
150 {
151 ClientConnectionFactory factory = h2;
152 SslContextFactory sslContextFactory = (SslContextFactory)context.get(SslClientConnectionFactory.SSL_CONTEXT_FACTORY_CONTEXT_KEY);
153 if (sslContextFactory != null)
154 {
155 ALPNClientConnectionFactory alpn = new ALPNClientConnectionFactory(getExecutor(), h2, getProtocols());
156 factory = new SslClientConnectionFactory(sslContextFactory, getByteBufferPool(), getExecutor(), alpn);
157 }
158 return factory.newConnection(endPoint, context);
159 });
160 }
161
162 if (selector == null)
163 {
164 selector = newSelectorManager();
165 addBean(selector);
166 }
167 selector.setConnectTimeout(getConnectTimeout());
168
169 super.doStart();
170 }
171
172 protected SelectorManager newSelectorManager()
173 {
174 return new ClientSelectorManager(getExecutor(), getScheduler(), getSelectors());
175 }
176
177 public Executor getExecutor()
178 {
179 return executor;
180 }
181
182 public void setExecutor(Executor executor)
183 {
184 this.updateBean(this.executor, executor);
185 this.executor = executor;
186 }
187
188 public Scheduler getScheduler()
189 {
190 return scheduler;
191 }
192
193 public void setScheduler(Scheduler scheduler)
194 {
195 this.updateBean(this.scheduler, scheduler);
196 this.scheduler = scheduler;
197 }
198
199 public ByteBufferPool getByteBufferPool()
200 {
201 return bufferPool;
202 }
203
204 public void setByteBufferPool(ByteBufferPool bufferPool)
205 {
206 this.updateBean(this.bufferPool, bufferPool);
207 this.bufferPool = bufferPool;
208 }
209
210 public ClientConnectionFactory getClientConnectionFactory()
211 {
212 return connectionFactory;
213 }
214
215 public void setClientConnectionFactory(ClientConnectionFactory connectionFactory)
216 {
217 this.updateBean(this.connectionFactory, connectionFactory);
218 this.connectionFactory = connectionFactory;
219 }
220
221 public FlowControlStrategy.Factory getFlowControlStrategyFactory()
222 {
223 return flowControlStrategyFactory;
224 }
225
226 public void setFlowControlStrategyFactory(FlowControlStrategy.Factory flowControlStrategyFactory)
227 {
228 this.flowControlStrategyFactory = flowControlStrategyFactory;
229 }
230
231 @ManagedAttribute("The number of selectors")
232 public int getSelectors()
233 {
234 return selectors;
235 }
236
237 public void setSelectors(int selectors)
238 {
239 this.selectors = selectors;
240 }
241
242 @ManagedAttribute("The idle timeout in milliseconds")
243 public long getIdleTimeout()
244 {
245 return idleTimeout;
246 }
247
248 public void setIdleTimeout(long idleTimeout)
249 {
250 this.idleTimeout = idleTimeout;
251 }
252
253 @ManagedAttribute("The connect timeout in milliseconds")
254 public long getConnectTimeout()
255 {
256 return connectTimeout;
257 }
258
259 public void setConnectTimeout(long connectTimeout)
260 {
261 this.connectTimeout = connectTimeout;
262 SelectorManager selector = this.selector;
263 if (selector != null)
264 selector.setConnectTimeout(connectTimeout);
265 }
266
267 @ManagedAttribute("The size of the buffer used to read from the network")
268 public int getInputBufferSize()
269 {
270 return inputBufferSize;
271 }
272
273 public void setInputBufferSize(int inputBufferSize)
274 {
275 this.inputBufferSize = inputBufferSize;
276 }
277
278 @ManagedAttribute("The ALPN protocol list")
279 public List<String> getProtocols()
280 {
281 return protocols;
282 }
283
284 public void setProtocols(List<String> protocols)
285 {
286 this.protocols = protocols;
287 }
288
289 @ManagedAttribute("The initial size of session's flow control receive window")
290 public int getInitialSessionRecvWindow()
291 {
292 return initialSessionRecvWindow;
293 }
294
295 public void setInitialSessionRecvWindow(int initialSessionRecvWindow)
296 {
297 this.initialSessionRecvWindow = initialSessionRecvWindow;
298 }
299
300 @ManagedAttribute("The initial size of stream's flow control receive window")
301 public int getInitialStreamRecvWindow()
302 {
303 return initialStreamRecvWindow;
304 }
305
306 public void setInitialStreamRecvWindow(int initialStreamRecvWindow)
307 {
308 this.initialStreamRecvWindow = initialStreamRecvWindow;
309 }
310
311 public void connect(InetSocketAddress address, Session.Listener listener, Promise<Session> promise)
312 {
313 connect(null, address, listener, promise);
314 }
315
316 public void connect(SslContextFactory sslContextFactory, InetSocketAddress address, Session.Listener listener, Promise<Session> promise)
317 {
318 connect(sslContextFactory, address, listener, promise, null);
319 }
320
321 public void connect(SslContextFactory sslContextFactory, InetSocketAddress address, Session.Listener listener, Promise<Session> promise, Map<String, Object> context)
322 {
323 try
324 {
325 SocketChannel channel = SocketChannel.open();
326 configure(channel);
327 channel.configureBlocking(false);
328 context = contextFrom(sslContextFactory, address, listener, promise, context);
329 if (channel.connect(address))
330 selector.accept(channel, context);
331 else
332 selector.connect(channel, context);
333 }
334 catch (Throwable x)
335 {
336 promise.failed(x);
337 }
338 }
339
340 public void accept(SslContextFactory sslContextFactory, SocketChannel channel, Session.Listener listener, Promise<Session> promise)
341 {
342 try
343 {
344 if (!channel.isConnected())
345 throw new IllegalStateException("SocketChannel must be connected");
346 channel.configureBlocking(false);
347 Map<String, Object> context = contextFrom(sslContextFactory, (InetSocketAddress)channel.getRemoteAddress(), listener, promise, null);
348 selector.accept(channel, context);
349 }
350 catch (Throwable x)
351 {
352 promise.failed(x);
353 }
354 }
355
356 private Map<String, Object> contextFrom(SslContextFactory sslContextFactory, InetSocketAddress address, Session.Listener listener, Promise<Session> promise, Map<String, Object> context)
357 {
358 if (context == null)
359 context = new HashMap<>();
360 context.put(HTTP2ClientConnectionFactory.CLIENT_CONTEXT_KEY, this);
361 context.put(HTTP2ClientConnectionFactory.SESSION_LISTENER_CONTEXT_KEY, listener);
362 context.put(HTTP2ClientConnectionFactory.SESSION_PROMISE_CONTEXT_KEY, promise);
363 if (sslContextFactory != null)
364 context.put(SslClientConnectionFactory.SSL_CONTEXT_FACTORY_CONTEXT_KEY, sslContextFactory);
365 context.put(SslClientConnectionFactory.SSL_PEER_HOST_CONTEXT_KEY, address.getHostString());
366 context.put(SslClientConnectionFactory.SSL_PEER_PORT_CONTEXT_KEY, address.getPort());
367 context.putIfAbsent(ClientConnectionFactory.CONNECTOR_CONTEXT_KEY, this);
368 return context;
369 }
370
371 protected void configure(SocketChannel channel) throws IOException
372 {
373 channel.socket().setTcpNoDelay(true);
374 }
375
376 private class ClientSelectorManager extends SelectorManager
377 {
378 private ClientSelectorManager(Executor executor, Scheduler scheduler, int selectors)
379 {
380 super(executor, scheduler, selectors);
381 }
382
383 @Override
384 protected EndPoint newEndPoint(SelectableChannel channel, ManagedSelector selector, SelectionKey selectionKey) throws IOException
385 {
386 SocketChannelEndPoint endp = new SocketChannelEndPoint(channel, selector, selectionKey, getScheduler());
387 endp.setIdleTimeout(getIdleTimeout());
388 return endp;
389 }
390
391 @Override
392 public Connection newConnection(SelectableChannel channel, EndPoint endpoint, Object attachment) throws IOException
393 {
394 @SuppressWarnings("unchecked")
395 Map<String, Object> context = (Map<String, Object>)attachment;
396 context.put(HTTP2ClientConnectionFactory.BYTE_BUFFER_POOL_CONTEXT_KEY, getByteBufferPool());
397 context.put(HTTP2ClientConnectionFactory.EXECUTOR_CONTEXT_KEY, getExecutor());
398 context.put(HTTP2ClientConnectionFactory.SCHEDULER_CONTEXT_KEY, getScheduler());
399 return getClientConnectionFactory().newConnection(endpoint, context);
400 }
401
402 @Override
403 protected void connectionFailed(SelectableChannel channel, Throwable failure, Object attachment)
404 {
405 @SuppressWarnings("unchecked")
406 Map<String, Object> context = (Map<String, Object>)attachment;
407 if (LOG.isDebugEnabled())
408 {
409 Object host = context.get(SslClientConnectionFactory.SSL_PEER_HOST_CONTEXT_KEY);
410 Object port = context.get(SslClientConnectionFactory.SSL_PEER_PORT_CONTEXT_KEY);
411 LOG.debug("Could not connect to {}:{}", host, port);
412 }
413 @SuppressWarnings("unchecked")
414 Promise<Session> promise = (Promise<Session>)context.get(HTTP2ClientConnectionFactory.SESSION_PROMISE_CONTEXT_KEY);
415 promise.failed(failure);
416 }
417 }
418 }