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.http2.examples;
28  
29  import java.io.IOException;
30  import java.net.URI;
31  import java.nio.ByteBuffer;
32  import java.util.List;
33  import java.util.concurrent.CountDownLatch;
34  import java.util.concurrent.TimeUnit;
35  
36  import org.apache.hc.core5.http.EntityDetails;
37  import org.apache.hc.core5.http.Header;
38  import org.apache.hc.core5.http.HttpConnection;
39  import org.apache.hc.core5.http.HttpException;
40  import org.apache.hc.core5.http.HttpResponse;
41  import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
42  import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
43  import org.apache.hc.core5.http.nio.AsyncRequestProducer;
44  import org.apache.hc.core5.http.nio.CapacityChannel;
45  import org.apache.hc.core5.http.nio.DataStreamChannel;
46  import org.apache.hc.core5.http.nio.RequestChannel;
47  import org.apache.hc.core5.http.nio.entity.StringAsyncEntityConsumer;
48  import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder;
49  import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
50  import org.apache.hc.core5.http.protocol.HttpContext;
51  import org.apache.hc.core5.http.protocol.HttpCoreContext;
52  import org.apache.hc.core5.http2.HttpVersionPolicy;
53  import org.apache.hc.core5.http2.config.H2Config;
54  import org.apache.hc.core5.http2.frame.RawFrame;
55  import org.apache.hc.core5.http2.impl.nio.H2StreamListener;
56  import org.apache.hc.core5.http2.impl.nio.bootstrap.H2RequesterBootstrap;
57  import org.apache.hc.core5.io.CloseMode;
58  import org.apache.hc.core5.reactor.IOReactorConfig;
59  import org.apache.hc.core5.util.Timeout;
60  
61  /**
62   * Example of full-duplex, streaming HTTP message exchanges with an asynchronous HTTP/2 requester.
63   */
64  public class H2FullDuplexClientExample {
65  
66      public static void main(final String[] args) throws Exception {
67  
68          final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
69                  .setSoTimeout(5, TimeUnit.SECONDS)
70                  .build();
71  
72          // Create and start requester
73          final H2Config h2Config = H2Config.custom()
74                  .setPushEnabled(false)
75                  .setMaxConcurrentStreams(100)
76                  .build();
77          final HttpAsyncRequester requester = H2RequesterBootstrap.bootstrap()
78                  .setIOReactorConfig(ioReactorConfig)
79                  .setH2Config(h2Config)
80                  .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_2)
81                  .setStreamListener(new H2StreamListener() {
82  
83                      @Override
84                      public void onHeaderInput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
85                          for (int i = 0; i < headers.size(); i++) {
86                              System.out.println(connection.getRemoteAddress() + " (" + streamId + ") << " + headers.get(i));
87                          }
88                      }
89  
90                      @Override
91                      public void onHeaderOutput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
92                          for (int i = 0; i < headers.size(); i++) {
93                              System.out.println(connection.getRemoteAddress() + " (" + streamId + ") >> " + headers.get(i));
94                          }
95                      }
96  
97                      @Override
98                      public void onFrameInput(final HttpConnection connection, final int streamId, final RawFrame frame) {
99                      }
100 
101                     @Override
102                     public void onFrameOutput(final HttpConnection connection, final int streamId, final RawFrame frame) {
103                     }
104 
105                     @Override
106                     public void onInputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
107                     }
108 
109                     @Override
110                     public void onOutputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
111                     }
112 
113                 })
114                 .create();
115 
116         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
117             System.out.println("HTTP requester shutting down");
118             requester.close(CloseMode.GRACEFUL);
119         }));
120         requester.start();
121 
122         final URI requestUri = new URI("http://nghttp2.org/httpbin/post");
123         final AsyncRequestProducer requestProducer = AsyncRequestBuilder.post(requestUri)
124                 .setEntity("stuff")
125                 .build();
126         final BasicResponseConsumer<String> responseConsumer = new BasicResponseConsumer<>(
127                 new StringAsyncEntityConsumer());
128 
129         final CountDownLatch latch = new CountDownLatch(1);
130         requester.execute(new AsyncClientExchangeHandler() {
131 
132             @Override
133             public void releaseResources() {
134                 requestProducer.releaseResources();
135                 responseConsumer.releaseResources();
136                 latch.countDown();
137             }
138 
139             @Override
140             public void cancel() {
141                 System.out.println(requestUri + " cancelled");
142             }
143 
144             @Override
145             public void failed(final Exception cause) {
146                 System.out.println(requestUri + "->" + cause);
147             }
148 
149             @Override
150             public void produceRequest(final RequestChannel channel, final HttpContext httpContext) throws HttpException, IOException {
151                 requestProducer.sendRequest(channel, httpContext);
152             }
153 
154             @Override
155             public int available() {
156                 return requestProducer.available();
157             }
158 
159             @Override
160             public void produce(final DataStreamChannel channel) throws IOException {
161                 requestProducer.produce(channel);
162             }
163 
164             @Override
165             public void consumeInformation(final HttpResponse response, final HttpContext httpContext) throws HttpException, IOException {
166                 System.out.println(requestUri + "->" + response.getCode());
167             }
168 
169             @Override
170             public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
171                 System.out.println(requestUri + "->" + response.getCode());
172                 responseConsumer.consumeResponse(response, entityDetails, httpContext, null);
173             }
174 
175             @Override
176             public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
177                 responseConsumer.updateCapacity(capacityChannel);
178             }
179 
180             @Override
181             public void consume(final ByteBuffer src) throws IOException {
182                 responseConsumer.consume(src);
183             }
184 
185             @Override
186             public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
187                 responseConsumer.streamEnd(trailers);
188             }
189 
190         }, Timeout.ofSeconds(30), HttpCoreContext.create());
191 
192         latch.await();
193         System.out.println("Shutting down I/O reactor");
194         requester.initiateShutdown();
195     }
196 
197 }