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 package org.apache.hc.client5.http.impl.async;
28
29 import java.io.IOException;
30 import java.io.InterruptedIOException;
31 import java.nio.ByteBuffer;
32 import java.util.List;
33 import java.util.concurrent.atomic.AtomicInteger;
34 import java.util.concurrent.atomic.AtomicReference;
35
36 import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
37 import org.apache.hc.client5.http.HttpRoute;
38 import org.apache.hc.client5.http.UserTokenHandler;
39 import org.apache.hc.client5.http.async.AsyncExecCallback;
40 import org.apache.hc.client5.http.async.AsyncExecChain;
41 import org.apache.hc.client5.http.async.AsyncExecChainHandler;
42 import org.apache.hc.client5.http.async.AsyncExecRuntime;
43 import org.apache.hc.client5.http.protocol.HttpClientContext;
44 import org.apache.hc.core5.annotation.Contract;
45 import org.apache.hc.core5.annotation.Internal;
46 import org.apache.hc.core5.annotation.ThreadingBehavior;
47 import org.apache.hc.core5.concurrent.CancellableDependency;
48 import org.apache.hc.core5.http.EntityDetails;
49 import org.apache.hc.core5.http.Header;
50 import org.apache.hc.core5.http.HttpException;
51 import org.apache.hc.core5.http.HttpRequest;
52 import org.apache.hc.core5.http.HttpResponse;
53 import org.apache.hc.core5.http.HttpStatus;
54 import org.apache.hc.core5.http.message.RequestLine;
55 import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
56 import org.apache.hc.core5.http.nio.AsyncDataConsumer;
57 import org.apache.hc.core5.http.nio.AsyncEntityProducer;
58 import org.apache.hc.core5.http.nio.CapacityChannel;
59 import org.apache.hc.core5.http.nio.DataStreamChannel;
60 import org.apache.hc.core5.http.nio.RequestChannel;
61 import org.apache.hc.core5.http.protocol.HttpContext;
62 import org.apache.hc.core5.http.protocol.HttpCoreContext;
63 import org.apache.hc.core5.http.protocol.HttpProcessor;
64 import org.apache.hc.core5.util.Args;
65 import org.apache.hc.core5.util.TimeValue;
66 import org.slf4j.Logger;
67 import org.slf4j.LoggerFactory;
68
69
70
71
72
73
74
75
76 @Contract(threading = ThreadingBehavior.STATELESS)
77 @Internal
78 class HttpAsyncMainClientExec implements AsyncExecChainHandler {
79
80 private static final Logger LOG = LoggerFactory.getLogger(HttpAsyncMainClientExec.class);
81
82 private final HttpProcessor httpProcessor;
83 private final ConnectionKeepAliveStrategy keepAliveStrategy;
84 private final UserTokenHandler userTokenHandler;
85
86 HttpAsyncMainClientExec(final HttpProcessor httpProcessor,
87 final ConnectionKeepAliveStrategy keepAliveStrategy,
88 final UserTokenHandler userTokenHandler) {
89 this.httpProcessor = Args.notNull(httpProcessor, "HTTP protocol processor");
90 this.keepAliveStrategy = keepAliveStrategy;
91 this.userTokenHandler = userTokenHandler;
92 }
93
94 @Override
95 public void execute(
96 final HttpRequest request,
97 final AsyncEntityProducer entityProducer,
98 final AsyncExecChain.Scope scope,
99 final AsyncExecChain chain,
100 final AsyncExecCallback asyncExecCallback) throws HttpException, IOException {
101 final String exchangeId = scope.exchangeId;
102 final HttpRoute route = scope.route;
103 final CancellableDependency operation = scope.cancellableDependency;
104 final HttpClientContext clientContext = scope.clientContext;
105 final AsyncExecRuntime execRuntime = scope.execRuntime;
106
107 if (LOG.isDebugEnabled()) {
108 LOG.debug("{} executing {}", exchangeId, new RequestLine(request));
109 }
110
111 final AtomicInteger messageCountDown = new AtomicInteger(2);
112 final AsyncClientExchangeHandler internalExchangeHandler = new AsyncClientExchangeHandler() {
113
114 private final AtomicReference<AsyncDataConsumer> entityConsumerRef = new AtomicReference<>();
115
116 @Override
117 public void releaseResources() {
118 final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
119 if (entityConsumer != null) {
120 entityConsumer.releaseResources();
121 }
122 }
123
124 @Override
125 public void failed(final Exception cause) {
126 final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
127 if (entityConsumer != null) {
128 entityConsumer.releaseResources();
129 }
130 execRuntime.markConnectionNonReusable();
131 asyncExecCallback.failed(cause);
132 }
133
134 @Override
135 public void cancel() {
136 if (messageCountDown.get() > 0) {
137 failed(new InterruptedIOException());
138 }
139 }
140
141 @Override
142 public void produceRequest(
143 final RequestChannel channel,
144 final HttpContext context) throws HttpException, IOException {
145
146 clientContext.setAttribute(HttpClientContext.HTTP_ROUTE, route);
147 clientContext.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
148 httpProcessor.process(request, entityProducer, clientContext);
149
150 channel.sendRequest(request, entityProducer, context);
151 if (entityProducer == null) {
152 messageCountDown.decrementAndGet();
153 }
154 }
155
156 @Override
157 public int available() {
158 return entityProducer.available();
159 }
160
161 @Override
162 public void produce(final DataStreamChannel channel) throws IOException {
163 entityProducer.produce(new DataStreamChannel() {
164
165 @Override
166 public void requestOutput() {
167 channel.requestOutput();
168 }
169
170 @Override
171 public int write(final ByteBuffer src) throws IOException {
172 return channel.write(src);
173 }
174
175 @Override
176 public void endStream(final List<? extends Header> trailers) throws IOException {
177 channel.endStream(trailers);
178 if (messageCountDown.decrementAndGet() <= 0) {
179 asyncExecCallback.completed();
180 }
181 }
182
183 @Override
184 public void endStream() throws IOException {
185 channel.endStream();
186 if (messageCountDown.decrementAndGet() <= 0) {
187 asyncExecCallback.completed();
188 }
189 }
190
191 });
192 }
193
194 @Override
195 public void consumeInformation(
196 final HttpResponse response,
197 final HttpContext context) throws HttpException, IOException {
198 asyncExecCallback.handleInformationResponse(response);
199 }
200
201 @Override
202 public void consumeResponse(
203 final HttpResponse response,
204 final EntityDetails entityDetails,
205 final HttpContext context) throws HttpException, IOException {
206
207 clientContext.setAttribute(HttpCoreContext.HTTP_RESPONSE, response);
208 httpProcessor.process(response, entityDetails, clientContext);
209
210 entityConsumerRef.set(asyncExecCallback.handleResponse(response, entityDetails));
211 if (response.getCode() >= HttpStatus.SC_CLIENT_ERROR) {
212 messageCountDown.decrementAndGet();
213 }
214 final TimeValue keepAliveDuration = keepAliveStrategy.getKeepAliveDuration(response, clientContext);
215 Object userToken = clientContext.getUserToken();
216 if (userToken == null) {
217 userToken = userTokenHandler.getUserToken(route, request, clientContext);
218 clientContext.setAttribute(HttpClientContext.USER_TOKEN, userToken);
219 }
220 execRuntime.markConnectionReusable(userToken, keepAliveDuration);
221 if (entityDetails == null) {
222 execRuntime.validateConnection();
223 if (messageCountDown.decrementAndGet() <= 0) {
224 asyncExecCallback.completed();
225 }
226 }
227 }
228
229 @Override
230 public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
231 final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
232 if (entityConsumer != null) {
233 entityConsumer.updateCapacity(capacityChannel);
234 } else {
235 capacityChannel.update(Integer.MAX_VALUE);
236 }
237 }
238
239 @Override
240 public void consume(final ByteBuffer src) throws IOException {
241 final AsyncDataConsumer entityConsumer = entityConsumerRef.get();
242 if (entityConsumer != null) {
243 entityConsumer.consume(src);
244 }
245 }
246
247 @Override
248 public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
249 final AsyncDataConsumer entityConsumer = entityConsumerRef.getAndSet(null);
250 if (entityConsumer != null) {
251 entityConsumer.streamEnd(trailers);
252 } else {
253 execRuntime.validateConnection();
254 }
255 if (messageCountDown.decrementAndGet() <= 0) {
256 asyncExecCallback.completed();
257 }
258 }
259
260 };
261
262 if (LOG.isDebugEnabled()) {
263 operation.setDependency(execRuntime.execute(
264 exchangeId,
265 new LoggingAsyncClientExchangeHandler(LOG, exchangeId, internalExchangeHandler),
266 clientContext));
267 } else {
268 operation.setDependency(execRuntime.execute(exchangeId, internalExchangeHandler, clientContext));
269 }
270 }
271
272 }