1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28 package org.apache.hc.core5.http2.impl.nio.bootstrap;
29
30 import java.io.IOException;
31 import java.net.InetSocketAddress;
32 import java.nio.ByteBuffer;
33 import java.util.List;
34 import java.util.Set;
35 import java.util.concurrent.Future;
36
37 import org.apache.hc.core5.annotation.Internal;
38 import org.apache.hc.core5.concurrent.Cancellable;
39 import org.apache.hc.core5.concurrent.CancellableDependency;
40 import org.apache.hc.core5.concurrent.ComplexFuture;
41 import org.apache.hc.core5.concurrent.FutureCallback;
42 import org.apache.hc.core5.function.Callback;
43 import org.apache.hc.core5.function.Decorator;
44 import org.apache.hc.core5.function.Resolver;
45 import org.apache.hc.core5.http.ConnectionClosedException;
46 import org.apache.hc.core5.http.EntityDetails;
47 import org.apache.hc.core5.http.Header;
48 import org.apache.hc.core5.http.HttpException;
49 import org.apache.hc.core5.http.HttpHost;
50 import org.apache.hc.core5.http.HttpRequest;
51 import org.apache.hc.core5.http.HttpResponse;
52 import org.apache.hc.core5.http.ProtocolException;
53 import org.apache.hc.core5.http.impl.DefaultAddressResolver;
54 import org.apache.hc.core5.http.impl.bootstrap.AsyncRequester;
55 import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
56 import org.apache.hc.core5.http.nio.AsyncPushConsumer;
57 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
58 import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
59 import org.apache.hc.core5.http.nio.CapacityChannel;
60 import org.apache.hc.core5.http.nio.DataStreamChannel;
61 import org.apache.hc.core5.http.nio.HandlerFactory;
62 import org.apache.hc.core5.http.nio.RequestChannel;
63 import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
64 import org.apache.hc.core5.http.nio.command.ShutdownCommand;
65 import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
66 import org.apache.hc.core5.http.nio.support.BasicClientExchangeHandler;
67 import org.apache.hc.core5.http.protocol.HttpContext;
68 import org.apache.hc.core5.http.protocol.HttpCoreContext;
69 import org.apache.hc.core5.http2.nio.pool.H2ConnPool;
70 import org.apache.hc.core5.net.URIAuthority;
71 import org.apache.hc.core5.reactor.Command;
72 import org.apache.hc.core5.reactor.IOEventHandlerFactory;
73 import org.apache.hc.core5.reactor.IOReactorConfig;
74 import org.apache.hc.core5.reactor.IOSession;
75 import org.apache.hc.core5.reactor.IOSessionListener;
76 import org.apache.hc.core5.util.Args;
77 import org.apache.hc.core5.util.TimeValue;
78 import org.apache.hc.core5.util.Timeout;
79
80
81
82
83
84
85 public class H2MultiplexingRequester extends AsyncRequester{
86
87 private final H2ConnPool connPool;
88
89
90
91
92 @Internal
93 public H2MultiplexingRequester(
94 final IOReactorConfig ioReactorConfig,
95 final IOEventHandlerFactory eventHandlerFactory,
96 final Decorator<IOSession> ioSessionDecorator,
97 final Callback<Exception> exceptionCallback,
98 final IOSessionListener sessionListener,
99 final Resolver<HttpHost, InetSocketAddress> addressResolver,
100 final TlsStrategy tlsStrategy) {
101 super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
102 ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE);
103 this.connPool = new H2ConnPool(this, addressResolver, tlsStrategy);
104 }
105
106 public void closeIdle(final TimeValue idleTime) {
107 connPool.closeIdle(idleTime);
108 }
109
110 public Set<HttpHost> getRoutes() {
111 return connPool.getRoutes();
112 }
113
114 public TimeValue getValidateAfterInactivity() {
115 return connPool.getValidateAfterInactivity();
116 }
117
118 public void setValidateAfterInactivity(final TimeValue timeValue) {
119 connPool.setValidateAfterInactivity(timeValue);
120 }
121
122 public Cancellable execute(
123 final AsyncClientExchangeHandler exchangeHandler,
124 final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
125 final Timeout timeout,
126 final HttpContext context) {
127 Args.notNull(exchangeHandler, "Exchange handler");
128 Args.notNull(timeout, "Timeout");
129 Args.notNull(context, "Context");
130 final CancellableExecution/CancellableExecution.html#CancellableExecution">CancellableExecution cancellableExecution = new CancellableExecution();
131 execute(exchangeHandler, pushHandlerFactory, cancellableExecution, timeout, context);
132 return cancellableExecution;
133 }
134
135 public Cancellable execute(
136 final AsyncClientExchangeHandler exchangeHandler,
137 final Timeout timeout,
138 final HttpContext context) {
139 return execute(exchangeHandler, null, timeout, context);
140 }
141
142 private void execute(
143 final AsyncClientExchangeHandler exchangeHandler,
144 final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
145 final CancellableDependency cancellableDependency,
146 final Timeout timeout,
147 final HttpContext context) {
148 Args.notNull(exchangeHandler, "Exchange handler");
149 Args.notNull(timeout, "Timeout");
150 Args.notNull(context, "Context");
151 try {
152 exchangeHandler.produceRequest(new RequestChannel() {
153
154 @Override
155 public void sendRequest(
156 final HttpRequest request,
157 final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
158 final String scheme = request.getScheme();
159 final URIAuthority authority = request.getAuthority();
160 if (authority == null) {
161 throw new ProtocolException("Request authority not specified");
162 }
163 final HttpHost5/http/HttpHost.html#HttpHost">HttpHost target = new HttpHost(scheme, authority);
164 connPool.getSession(target, timeout, new FutureCallback<IOSession>() {
165
166 @Override
167 public void completed(final IOSession ioSession) {
168 ioSession.enqueue(new RequestExecutionCommand(new AsyncClientExchangeHandler() {
169
170 @Override
171 public void releaseResources() {
172 exchangeHandler.releaseResources();
173 }
174
175 @Override
176 public void produceRequest(final RequestChannel channel, final HttpContext httpContext) throws HttpException, IOException {
177 channel.sendRequest(request, entityDetails, httpContext);
178 }
179
180 @Override
181 public int available() {
182 return exchangeHandler.available();
183 }
184
185 @Override
186 public void produce(final DataStreamChannel channel) throws IOException {
187 exchangeHandler.produce(channel);
188 }
189
190 @Override
191 public void consumeInformation(final HttpResponse response, final HttpContext httpContext) throws HttpException, IOException {
192 exchangeHandler.consumeInformation(response, httpContext);
193 }
194
195 @Override
196 public void consumeResponse(
197 final HttpResponse response, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
198 exchangeHandler.consumeResponse(response, entityDetails, httpContext);
199 }
200
201 @Override
202 public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
203 exchangeHandler.updateCapacity(capacityChannel);
204 }
205
206 @Override
207 public void consume(final ByteBuffer src) throws IOException {
208 exchangeHandler.consume(src);
209 }
210
211 @Override
212 public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
213 exchangeHandler.streamEnd(trailers);
214 }
215
216 @Override
217 public void cancel() {
218 exchangeHandler.cancel();
219 }
220
221 @Override
222 public void failed(final Exception cause) {
223 exchangeHandler.failed(cause);
224 }
225
226 }, pushHandlerFactory, cancellableDependency, context), Command.Priority.NORMAL);
227 if (!ioSession.isOpen()) {
228 exchangeHandler.failed(new ConnectionClosedException());
229 }
230 }
231
232 @Override
233 public void failed(final Exception ex) {
234 exchangeHandler.failed(ex);
235 }
236
237 @Override
238 public void cancelled() {
239 exchangeHandler.cancel();
240 }
241
242 });
243
244 }
245
246 }, context);
247 } catch (final IOException | HttpException ex) {
248 exchangeHandler.failed(ex);
249 }
250 }
251
252 public final <T> Future<T> execute(
253 final AsyncRequestProducer requestProducer,
254 final AsyncResponseConsumer<T> responseConsumer,
255 final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
256 final Timeout timeout,
257 final HttpContext context,
258 final FutureCallback<T> callback) {
259 Args.notNull(requestProducer, "Request producer");
260 Args.notNull(responseConsumer, "Response consumer");
261 Args.notNull(timeout, "Timeout");
262 final ComplexFuture<T> future = new ComplexFuture<>(callback);
263 final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(requestProducer, responseConsumer, new FutureCallback<T>() {
264
265 @Override
266 public void completed(final T result) {
267 future.completed(result);
268 }
269
270 @Override
271 public void failed(final Exception ex) {
272 future.failed(ex);
273 }
274
275 @Override
276 public void cancelled() {
277 future.cancel();
278 }
279
280 });
281 execute(exchangeHandler, pushHandlerFactory, future, timeout, context != null ? context : HttpCoreContext.create());
282 return future;
283 }
284
285 public final <T> Future<T> execute(
286 final AsyncRequestProducer requestProducer,
287 final AsyncResponseConsumer<T> responseConsumer,
288 final Timeout timeout,
289 final HttpContext context,
290 final FutureCallback<T> callback) {
291 return execute(requestProducer, responseConsumer, null, timeout, context, callback);
292 }
293
294 public final <T> Future<T> execute(
295 final AsyncRequestProducer requestProducer,
296 final AsyncResponseConsumer<T> responseConsumer,
297 final Timeout timeout,
298 final FutureCallback<T> callback) {
299 return execute(requestProducer, responseConsumer, null, timeout, null, callback);
300 }
301
302 }