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  import org.apache.hc.core5.util.Timeout;
57  
58  class ClientHttp1StreamHandler implements ResourceHolder {
59  
60      private final Http1StreamChannel<HttpRequest> outputChannel;
61      private final DataStreamChannel internalDataChannel;
62      private final HttpProcessor httpProcessor;
63      private final H1Config h1Config;
64      private final ConnectionReuseStrategy connectionReuseStrategy;
65      private final AsyncClientExchangeHandler exchangeHandler;
66      private final HttpCoreContext context;
67      private final AtomicBoolean requestCommitted;
68      private final AtomicBoolean done;
69  
70      private volatile boolean keepAlive;
71      private volatile Timeout timeout;
72      private volatile HttpRequest committedRequest;
73      private volatile MessageState requestState;
74      private volatile MessageState responseState;
75  
76      ClientHttp1StreamHandler(
77              final Http1StreamChannel<HttpRequest> outputChannel,
78              final HttpProcessor httpProcessor,
79              final H1Config h1Config,
80              final ConnectionReuseStrategy connectionReuseStrategy,
81              final AsyncClientExchangeHandler exchangeHandler,
82              final HttpCoreContext context) {
83          this.outputChannel = outputChannel;
84          this.internalDataChannel = new DataStreamChannel() {
85  
86              @Override
87              public void requestOutput() {
88                  outputChannel.requestOutput();
89              }
90  
91              @Override
92              public void endStream(final List<? extends Header> trailers) throws IOException {
93                  outputChannel.complete(trailers);
94                  requestState = MessageState.COMPLETE;
95              }
96  
97              @Override
98              public int write(final ByteBuffer src) throws IOException {
99                  return outputChannel.write(src);
100             }
101 
102             @Override
103             public void endStream() throws IOException {
104                 endStream(null);
105             }
106 
107         };
108 
109         this.httpProcessor = httpProcessor;
110         this.h1Config = h1Config;
111         this.connectionReuseStrategy = connectionReuseStrategy;
112         this.exchangeHandler = exchangeHandler;
113         this.context = context;
114         this.requestCommitted = new AtomicBoolean(false);
115         this.done = new AtomicBoolean(false);
116         this.keepAlive = true;
117         this.requestState = MessageState.IDLE;
118         this.responseState = MessageState.HEADERS;
119     }
120 
121     boolean isResponseFinal() {
122         return responseState == MessageState.COMPLETE;
123     }
124 
125     boolean isCompleted() {
126         return requestState == MessageState.COMPLETE && responseState == MessageState.COMPLETE;
127     }
128 
129     String getRequestMethod() {
130         return committedRequest != null ? committedRequest.getMethod() : null;
131     }
132 
133     boolean isOutputReady() {
134         switch (requestState) {
135             case IDLE:
136             case ACK:
137                 return true;
138             case BODY:
139                 return exchangeHandler.available() > 0;
140             default:
141                 return false;
142         }
143     }
144 
145     private void commitRequest(final HttpRequest request, final EntityDetails entityDetails) throws IOException, HttpException {
146         if (requestCommitted.compareAndSet(false, true)) {
147             final ProtocolVersion transportVersion = request.getVersion();
148             if (transportVersion != null && transportVersion.greaterEquals(HttpVersion.HTTP_2)) {
149                 throw new UnsupportedHttpVersionException(transportVersion);
150             }
151             context.setProtocolVersion(transportVersion);
152             context.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
153 
154             httpProcessor.process(request, entityDetails, context);
155 
156             final boolean endStream = entityDetails == null;
157             if (endStream) {
158                 outputChannel.submit(request, endStream, FlushMode.IMMEDIATE);
159                 committedRequest = request;
160                 requestState = MessageState.COMPLETE;
161             } else {
162                 final Header h = request.getFirstHeader(HttpHeaders.EXPECT);
163                 final boolean expectContinue = h != null && "100-continue".equalsIgnoreCase(h.getValue());
164                 outputChannel.submit(request, endStream, expectContinue ? FlushMode.IMMEDIATE : FlushMode.BUFFER);
165                 committedRequest = request;
166                 if (expectContinue) {
167                     requestState = MessageState.ACK;
168                     timeout = outputChannel.getSocketTimeout();
169                     outputChannel.setSocketTimeout(h1Config.getWaitForContinueTimeout());
170                 } else {
171                     requestState = MessageState.BODY;
172                     exchangeHandler.produce(internalDataChannel);
173                 }
174             }
175         } else {
176             throw new HttpException("Request already committed");
177         }
178     }
179 
180     void produceOutput() throws HttpException, IOException {
181         switch (requestState) {
182             case IDLE:
183                 requestState = MessageState.HEADERS;
184                 exchangeHandler.produceRequest(new RequestChannel() {
185 
186                     @Override
187                     public void sendRequest(
188                             final HttpRequest request, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
189                         commitRequest(request, entityDetails);
190                     }
191 
192                 }, context);
193                 break;
194             case ACK:
195                 outputChannel.suspendOutput();
196                 break;
197             case BODY:
198                 exchangeHandler.produce(internalDataChannel);
199                 break;
200         }
201     }
202 
203     void consumeHeader(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException {
204         if (done.get() || responseState != MessageState.HEADERS) {
205             throw new ProtocolException("Unexpected message head");
206         }
207         final ProtocolVersion transportVersion = response.getVersion();
208         if (transportVersion != null && transportVersion.greaterEquals(HttpVersion.HTTP_2)) {
209             throw new UnsupportedHttpVersionException(transportVersion);
210         }
211 
212         final int status = response.getCode();
213         if (status < HttpStatus.SC_INFORMATIONAL) {
214             throw new ProtocolException("Invalid response: " + new StatusLine(response));
215         }
216         if (status > HttpStatus.SC_CONTINUE && status < HttpStatus.SC_SUCCESS) {
217             exchangeHandler.consumeInformation(response, context);
218         } else {
219             if (!connectionReuseStrategy.keepAlive(committedRequest, response, context)) {
220                 keepAlive = false;
221             }
222         }
223         if (requestState == MessageState.ACK) {
224             if (status == HttpStatus.SC_CONTINUE || status >= HttpStatus.SC_SUCCESS) {
225                 outputChannel.setSocketTimeout(timeout);
226                 requestState = MessageState.BODY;
227                 if (status < HttpStatus.SC_CLIENT_ERROR) {
228                     exchangeHandler.produce(internalDataChannel);
229                 }
230             }
231         }
232         if (status < HttpStatus.SC_SUCCESS) {
233             return;
234         }
235         if (requestState == MessageState.BODY) {
236             if (keepAlive && status >= HttpStatus.SC_CLIENT_ERROR) {
237                 requestState = MessageState.COMPLETE;
238                 if (!outputChannel.abortGracefully()) {
239                     keepAlive = false;
240                 }
241             }
242         }
243 
244         context.setAttribute(HttpCoreContext.HTTP_RESPONSE, response);
245         httpProcessor.process(response, entityDetails, context);
246 
247         exchangeHandler.consumeResponse(response, entityDetails, context);
248         if (entityDetails == null) {
249             if (!keepAlive) {
250                 outputChannel.close();
251             }
252             responseState = MessageState.COMPLETE;
253         } else {
254             responseState = MessageState.BODY;
255         }
256     }
257 
258     int consumeData(final ByteBuffer src) throws HttpException, IOException {
259         if (done.get() || responseState != MessageState.BODY) {
260             throw new ProtocolException("Unexpected message data");
261         }
262         return exchangeHandler.consume(src);
263     }
264 
265     void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
266         exchangeHandler.updateCapacity(capacityChannel);
267     }
268 
269     void dataEnd(final List<? extends Header> trailers) throws HttpException, IOException {
270         if (done.get() || responseState != MessageState.BODY) {
271             throw new ProtocolException("Unexpected message data");
272         }
273         if (!keepAlive) {
274             outputChannel.close();
275         }
276         responseState = MessageState.COMPLETE;
277         exchangeHandler.streamEnd(trailers);
278     }
279 
280     boolean handleTimeout() {
281         if (requestState == MessageState.ACK) {
282             requestState = MessageState.BODY;
283             outputChannel.setSocketTimeout(timeout);
284             outputChannel.requestOutput();
285             return true;
286         }
287         return false;
288     }
289 
290     void failed(final Exception cause) {
291         if (!done.get()) {
292             exchangeHandler.failed(cause);
293         }
294     }
295 
296     @Override
297     public void releaseResources() {
298         if (done.compareAndSet(false, true)) {
299             responseState = MessageState.COMPLETE;
300             requestState = MessageState.COMPLETE;
301             exchangeHandler.releaseResources();
302         }
303     }
304 
305     @Override
306     public String toString() {
307         return "[" +
308                 "requestState=" + requestState +
309                 ", responseState=" + responseState +
310                 ']';
311     }
312 
313 }
314