View Javadoc
1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  package org.apache.hc.core5.http.impl.nio;
28  
29  import java.io.IOException;
30  import java.nio.ByteBuffer;
31  import java.util.List;
32  import java.util.concurrent.atomic.AtomicBoolean;
33  
34  import org.apache.hc.core5.http.ConnectionReuseStrategy;
35  import org.apache.hc.core5.http.EntityDetails;
36  import org.apache.hc.core5.http.Header;
37  import org.apache.hc.core5.http.HttpException;
38  import org.apache.hc.core5.http.HttpHeaders;
39  import org.apache.hc.core5.http.HttpRequest;
40  import org.apache.hc.core5.http.HttpResponse;
41  import org.apache.hc.core5.http.HttpStatus;
42  import org.apache.hc.core5.http.HttpVersion;
43  import org.apache.hc.core5.http.ProtocolException;
44  import org.apache.hc.core5.http.ProtocolVersion;
45  import org.apache.hc.core5.http.UnsupportedHttpVersionException;
46  import org.apache.hc.core5.http.config.H1Config;
47  import org.apache.hc.core5.http.message.StatusLine;
48  import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
49  import org.apache.hc.core5.http.nio.CapacityChannel;
50  import org.apache.hc.core5.http.nio.DataStreamChannel;
51  import org.apache.hc.core5.http.nio.RequestChannel;
52  import org.apache.hc.core5.http.nio.ResourceHolder;
53  import org.apache.hc.core5.http.protocol.HttpContext;
54  import org.apache.hc.core5.http.protocol.HttpCoreContext;
55  import org.apache.hc.core5.http.protocol.HttpProcessor;
56  
57  class ClientHttp1StreamHandler implements ResourceHolder {
58  
59      private final Http1StreamChannel<HttpRequest> outputChannel;
60      private final DataStreamChannel internalDataChannel;
61      private final HttpProcessor httpProcessor;
62      private final H1Config h1Config;
63      private final ConnectionReuseStrategy connectionReuseStrategy;
64      private final AsyncClientExchangeHandler exchangeHandler;
65      private final HttpCoreContext context;
66      private final AtomicBoolean requestCommitted;
67      private final AtomicBoolean done;
68  
69      private volatile boolean keepAlive;
70      private volatile int timeout;
71      private volatile HttpRequest committedRequest;
72      private volatile MessageState requestState;
73      private volatile MessageState responseState;
74  
75      ClientHttp1StreamHandler(
76              final Http1StreamChannel<HttpRequest> outputChannel,
77              final HttpProcessor httpProcessor,
78              final H1Config h1Config,
79              final ConnectionReuseStrategy connectionReuseStrategy,
80              final AsyncClientExchangeHandler exchangeHandler,
81              final HttpCoreContext context) {
82          this.outputChannel = outputChannel;
83          this.internalDataChannel = new DataStreamChannel() {
84  
85              @Override
86              public void requestOutput() {
87                  outputChannel.requestOutput();
88              }
89  
90              @Override
91              public void endStream(final List<? extends Header> trailers) throws IOException {
92                  outputChannel.complete(trailers);
93                  requestState = MessageState.COMPLETE;
94              }
95  
96              @Override
97              public int write(final ByteBuffer src) throws IOException {
98                  return outputChannel.write(src);
99              }
100 
101             @Override
102             public void endStream() throws IOException {
103                 endStream(null);
104             }
105 
106         };
107 
108         this.httpProcessor = httpProcessor;
109         this.h1Config = h1Config;
110         this.connectionReuseStrategy = connectionReuseStrategy;
111         this.exchangeHandler = exchangeHandler;
112         this.context = context;
113         this.requestCommitted = new AtomicBoolean(false);
114         this.done = new AtomicBoolean(false);
115         this.keepAlive = true;
116         this.requestState = MessageState.IDLE;
117         this.responseState = MessageState.HEADERS;
118     }
119 
120     boolean isResponseFinal() {
121         return responseState == MessageState.COMPLETE;
122     }
123 
124     boolean isCompleted() {
125         return requestState == MessageState.COMPLETE && responseState == MessageState.COMPLETE;
126     }
127 
128     String getRequestMethod() {
129         return committedRequest != null ? committedRequest.getMethod() : null;
130     }
131 
132     boolean isOutputReady() {
133         switch (requestState) {
134             case IDLE:
135             case ACK:
136                 return true;
137             case BODY:
138                 return exchangeHandler.available() > 0;
139             default:
140                 return false;
141         }
142     }
143 
144     private void commitRequest(final HttpRequest request, final EntityDetails entityDetails) throws IOException, HttpException {
145         if (requestCommitted.compareAndSet(false, true)) {
146             final ProtocolVersion transportVersion = request.getVersion();
147             if (transportVersion != null && transportVersion.greaterEquals(HttpVersion.HTTP_2)) {
148                 throw new UnsupportedHttpVersionException(transportVersion);
149             }
150             context.setProtocolVersion(transportVersion);
151             context.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
152 
153             httpProcessor.process(request, entityDetails, context);
154 
155             final boolean endStream = entityDetails == null;
156             outputChannel.submit(request, endStream);
157             committedRequest = request;
158             if (endStream) {
159                 requestState = MessageState.COMPLETE;
160             } else {
161                 final Header h = request.getFirstHeader(HttpHeaders.EXPECT);
162                 final boolean expectContinue = h != null && "100-continue".equalsIgnoreCase(h.getValue());
163                 if (expectContinue) {
164                     requestState = MessageState.ACK;
165                     timeout = outputChannel.getSocketTimeoutMillis();
166                     outputChannel.setSocketTimeoutMillis(h1Config.getWaitForContinueTimeoutMillis());
167                 } else {
168                     requestState = MessageState.BODY;
169                     exchangeHandler.produce(internalDataChannel);
170                 }
171             }
172         } else {
173             throw new HttpException("Request already committed");
174         }
175     }
176 
177     void produceOutput() throws HttpException, IOException {
178         switch (requestState) {
179             case IDLE:
180                 requestState = MessageState.HEADERS;
181                 exchangeHandler.produceRequest(new RequestChannel() {
182 
183                     @Override
184                     public void sendRequest(
185                             final HttpRequest request, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
186                         commitRequest(request, entityDetails);
187                     }
188 
189                 }, context);
190                 break;
191             case ACK:
192                 outputChannel.suspendOutput();
193                 break;
194             case BODY:
195                 exchangeHandler.produce(internalDataChannel);
196                 break;
197         }
198     }
199 
200     void consumeHeader(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException {
201         if (done.get() || responseState != MessageState.HEADERS) {
202             throw new ProtocolException("Unexpected message head");
203         }
204         final ProtocolVersion transportVersion = response.getVersion();
205         if (transportVersion != null && transportVersion.greaterEquals(HttpVersion.HTTP_2)) {
206             throw new UnsupportedHttpVersionException(transportVersion);
207         }
208 
209         final int status = response.getCode();
210         if (status < HttpStatus.SC_INFORMATIONAL) {
211             throw new ProtocolException("Invalid response: " + new StatusLine(response));
212         }
213         if (status > HttpStatus.SC_CONTINUE && status < HttpStatus.SC_SUCCESS) {
214             exchangeHandler.consumeInformation(response, context);
215         } else {
216             if (!connectionReuseStrategy.keepAlive(committedRequest, response, context)) {
217                 keepAlive = false;
218             }
219         }
220         if (requestState == MessageState.ACK) {
221             if (status == HttpStatus.SC_CONTINUE || status >= HttpStatus.SC_SUCCESS) {
222                 outputChannel.setSocketTimeoutMillis(timeout);
223                 requestState = MessageState.BODY;
224                 if (status < HttpStatus.SC_CLIENT_ERROR) {
225                     exchangeHandler.produce(internalDataChannel);
226                 }
227             }
228         }
229         if (status < HttpStatus.SC_SUCCESS) {
230             return;
231         }
232         if (requestState == MessageState.BODY) {
233             if (keepAlive && status >= HttpStatus.SC_CLIENT_ERROR) {
234                 requestState = MessageState.COMPLETE;
235                 if (!outputChannel.abortGracefully()) {
236                     keepAlive = false;
237                 }
238             }
239         }
240 
241         context.setAttribute(HttpCoreContext.HTTP_RESPONSE, response);
242         httpProcessor.process(response, entityDetails, context);
243 
244         exchangeHandler.consumeResponse(response, entityDetails, context);
245         if (entityDetails == null) {
246             if (!keepAlive) {
247                 outputChannel.close();
248             }
249             responseState = MessageState.COMPLETE;
250         } else {
251             responseState = MessageState.BODY;
252         }
253     }
254 
255     int consumeData(final ByteBuffer src) throws HttpException, IOException {
256         if (done.get() || responseState != MessageState.BODY) {
257             throw new ProtocolException("Unexpected message data");
258         }
259         return exchangeHandler.consume(src);
260     }
261 
262     void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
263         exchangeHandler.updateCapacity(capacityChannel);
264     }
265 
266     void dataEnd(final List<? extends Header> trailers) throws HttpException, IOException {
267         if (done.get() || responseState != MessageState.BODY) {
268             throw new ProtocolException("Unexpected message data");
269         }
270         if (!keepAlive) {
271             outputChannel.close();
272         }
273         responseState = MessageState.COMPLETE;
274         exchangeHandler.streamEnd(trailers);
275     }
276 
277     boolean handleTimeout() {
278         if (requestState == MessageState.ACK) {
279             requestState = MessageState.BODY;
280             outputChannel.setSocketTimeoutMillis(timeout);
281             outputChannel.requestOutput();
282             return true;
283         }
284         return false;
285     }
286 
287     void failed(final Exception cause) {
288         if (!done.get()) {
289             exchangeHandler.failed(cause);
290         }
291     }
292 
293     @Override
294     public void releaseResources() {
295         if (done.compareAndSet(false, true)) {
296             responseState = MessageState.COMPLETE;
297             requestState = MessageState.COMPLETE;
298             exchangeHandler.releaseResources();
299         }
300     }
301 
302     @Override
303     public String toString() {
304         return "[" +
305                 "requestState=" + requestState +
306                 ", responseState=" + responseState +
307                 ']';
308     }
309 
310 }
311