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.concurrent.FutureContribution;
43 import org.apache.hc.core5.function.Callback;
44 import org.apache.hc.core5.function.Decorator;
45 import org.apache.hc.core5.function.Resolver;
46 import org.apache.hc.core5.http.ConnectionClosedException;
47 import org.apache.hc.core5.http.EntityDetails;
48 import org.apache.hc.core5.http.Header;
49 import org.apache.hc.core5.http.HttpException;
50 import org.apache.hc.core5.http.HttpHost;
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 = 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((request, entityDetails, httpContext) -> {
153 final String scheme = request.getScheme();
154 final URIAuthority authority = request.getAuthority();
155 if (authority == null) {
156 throw new ProtocolException("Request authority not specified");
157 }
158 final HttpHost target = new HttpHost(scheme, authority);
159 connPool.getSession(target, timeout, new FutureCallback<IOSession>() {
160
161 @Override
162 public void completed(final IOSession ioSession) {
163 ioSession.enqueue(new RequestExecutionCommand(new AsyncClientExchangeHandler() {
164
165 @Override
166 public void releaseResources() {
167 exchangeHandler.releaseResources();
168 }
169
170 @Override
171 public void produceRequest(final RequestChannel channel, final HttpContext httpContext) throws HttpException, IOException {
172 channel.sendRequest(request, entityDetails, httpContext);
173 }
174
175 @Override
176 public int available() {
177 return exchangeHandler.available();
178 }
179
180 @Override
181 public void produce(final DataStreamChannel channel) throws IOException {
182 exchangeHandler.produce(channel);
183 }
184
185 @Override
186 public void consumeInformation(final HttpResponse response, final HttpContext httpContext) throws HttpException, IOException {
187 exchangeHandler.consumeInformation(response, httpContext);
188 }
189
190 @Override
191 public void consumeResponse(
192 final HttpResponse response, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
193 exchangeHandler.consumeResponse(response, entityDetails, httpContext);
194 }
195
196 @Override
197 public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
198 exchangeHandler.updateCapacity(capacityChannel);
199 }
200
201 @Override
202 public void consume(final ByteBuffer src) throws IOException {
203 exchangeHandler.consume(src);
204 }
205
206 @Override
207 public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
208 exchangeHandler.streamEnd(trailers);
209 }
210
211 @Override
212 public void cancel() {
213 exchangeHandler.cancel();
214 }
215
216 @Override
217 public void failed(final Exception cause) {
218 exchangeHandler.failed(cause);
219 }
220
221 }, pushHandlerFactory, cancellableDependency, context), Command.Priority.NORMAL);
222 if (!ioSession.isOpen()) {
223 exchangeHandler.failed(new ConnectionClosedException());
224 }
225 }
226
227 @Override
228 public void failed(final Exception ex) {
229 exchangeHandler.failed(ex);
230 }
231
232 @Override
233 public void cancelled() {
234 exchangeHandler.cancel();
235 }
236
237 });
238
239 }, context);
240 } catch (final IOException | HttpException ex) {
241 exchangeHandler.failed(ex);
242 }
243 }
244
245 public final <T> Future<T> execute(
246 final AsyncRequestProducer requestProducer,
247 final AsyncResponseConsumer<T> responseConsumer,
248 final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
249 final Timeout timeout,
250 final HttpContext context,
251 final FutureCallback<T> callback) {
252 Args.notNull(requestProducer, "Request producer");
253 Args.notNull(responseConsumer, "Response consumer");
254 Args.notNull(timeout, "Timeout");
255 final ComplexFuture<T> future = new ComplexFuture<>(callback);
256 final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(
257 requestProducer,
258 responseConsumer,
259 new FutureContribution<T>(future) {
260
261 @Override
262 public void completed(final T result) {
263 future.completed(result);
264 }
265
266 });
267 execute(exchangeHandler, pushHandlerFactory, future, timeout, context != null ? context : HttpCoreContext.create());
268 return future;
269 }
270
271 public final <T> Future<T> execute(
272 final AsyncRequestProducer requestProducer,
273 final AsyncResponseConsumer<T> responseConsumer,
274 final Timeout timeout,
275 final HttpContext context,
276 final FutureCallback<T> callback) {
277 return execute(requestProducer, responseConsumer, null, timeout, context, callback);
278 }
279
280 public final <T> Future<T> execute(
281 final AsyncRequestProducer requestProducer,
282 final AsyncResponseConsumer<T> responseConsumer,
283 final Timeout timeout,
284 final FutureCallback<T> callback) {
285 return execute(requestProducer, responseConsumer, null, timeout, null, callback);
286 }
287
288 @Internal
289 public H2ConnPool getConnPool() {
290 return connPool;
291 }
292
293 }