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.testing.async;
28
29 import java.util.Arrays;
30 import java.util.Collection;
31 import java.util.LinkedList;
32 import java.util.Queue;
33 import java.util.Random;
34 import java.util.concurrent.Future;
35 import java.util.concurrent.TimeUnit;
36
37 import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
38 import org.apache.hc.client5.http.impl.async.MinimalHttpAsyncClient;
39 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
40 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
41 import org.apache.hc.client5.http.protocol.HttpClientContext;
42 import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
43 import org.apache.hc.client5.testing.SSLTestContexts;
44 import org.apache.hc.core5.http.ContentType;
45 import org.apache.hc.core5.http.HttpHost;
46 import org.apache.hc.core5.http.HttpResponse;
47 import org.apache.hc.core5.http.HttpVersion;
48 import org.apache.hc.core5.http.Message;
49 import org.apache.hc.core5.http.Method;
50 import org.apache.hc.core5.http.URIScheme;
51 import org.apache.hc.core5.http.config.Http1Config;
52 import org.apache.hc.core5.http.nio.AsyncClientEndpoint;
53 import org.apache.hc.core5.http.nio.entity.AsyncEntityProducers;
54 import org.apache.hc.core5.http.nio.entity.BasicAsyncEntityConsumer;
55 import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
56 import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
57 import org.apache.hc.core5.http2.HttpVersionPolicy;
58 import org.apache.hc.core5.http2.config.H2Config;
59 import org.apache.hc.core5.reactor.IOReactorConfig;
60 import org.hamcrest.CoreMatchers;
61 import org.junit.Assert;
62 import org.junit.Test;
63 import org.junit.runner.RunWith;
64 import org.junit.runners.Parameterized;
65
66 @RunWith(Parameterized.class)
67 public class TestHttpAsyncMinimal extends AbstractHttpAsyncFundamentalsTest<MinimalHttpAsyncClient> {
68
69 @Parameterized.Parameters(name = "Minimal {0} {1}")
70 public static Collection<Object[]> protocols() {
71 return Arrays.asList(new Object[][]{
72 { HttpVersion.HTTP_1_1, URIScheme.HTTP },
73 { HttpVersion.HTTP_1_1, URIScheme.HTTPS },
74 { HttpVersion.HTTP_2, URIScheme.HTTP },
75 { HttpVersion.HTTP_2, URIScheme.HTTPS }
76 });
77 }
78
79 protected final HttpVersion version;
80
81 public TestHttpAsyncMinimal(final HttpVersion version, final URIScheme scheme) {
82 super(scheme);
83 this.version = version;
84 }
85
86 @Override
87 protected MinimalHttpAsyncClient createClient() throws Exception {
88 final PoolingAsyncClientConnectionManager connectionManager = PoolingAsyncClientConnectionManagerBuilder.create()
89 .setTlsStrategy(new DefaultClientTlsStrategy(SSLTestContexts.createClientSSLContext()))
90 .build();
91 final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
92 .setSoTimeout(TIMEOUT)
93 .build();
94 if (version.greaterEquals(HttpVersion.HTTP_2)) {
95 return HttpAsyncClients.createMinimal(
96 HttpVersionPolicy.FORCE_HTTP_2, H2Config.DEFAULT, Http1Config.DEFAULT, ioReactorConfig, connectionManager);
97 } else {
98 return HttpAsyncClients.createMinimal(
99 HttpVersionPolicy.FORCE_HTTP_1, H2Config.DEFAULT, Http1Config.DEFAULT, ioReactorConfig, connectionManager);
100 }
101 }
102
103 @Override
104 public HttpHost start() throws Exception {
105 if (version.greaterEquals(HttpVersion.HTTP_2)) {
106 return super.start(null, H2Config.DEFAULT);
107 } else {
108 return super.start(null, Http1Config.DEFAULT);
109 }
110 }
111
112 @Test
113 public void testConcurrentPostRequestsSameEndpoint() throws Exception {
114 final HttpHost target = start();
115 final byte[] b1 = new byte[1024];
116 final Random rnd = new Random(System.currentTimeMillis());
117 rnd.nextBytes(b1);
118
119 final int reqCount = 20;
120
121 final Future<AsyncClientEndpoint> endpointLease = httpclient.lease(target, null);
122 final AsyncClientEndpoint endpoint = endpointLease.get(5, TimeUnit.SECONDS);
123 try {
124 final Queue<Future<Message<HttpResponse, byte[]>>> queue = new LinkedList<>();
125 for (int i = 0; i < reqCount; i++) {
126 final Future<Message<HttpResponse, byte[]>> future = endpoint.execute(
127 new BasicRequestProducer(Method.GET, target, "/echo/",
128 AsyncEntityProducers.create(b1, ContentType.APPLICATION_OCTET_STREAM)),
129 new BasicResponseConsumer<>(new BasicAsyncEntityConsumer()), HttpClientContext.create(), null);
130 queue.add(future);
131 }
132 while (!queue.isEmpty()) {
133 final Future<Message<HttpResponse, byte[]>> future = queue.remove();
134 final Message<HttpResponse, byte[]> responseMessage = future.get();
135 Assert.assertThat(responseMessage, CoreMatchers.notNullValue());
136 final HttpResponse response = responseMessage.getHead();
137 Assert.assertThat(response.getCode(), CoreMatchers.equalTo(200));
138 final byte[] b2 = responseMessage.getBody();
139 Assert.assertThat(b1, CoreMatchers.equalTo(b2));
140 endpoint.releaseAndReuse();
141 }
142 } finally {
143 endpoint.releaseAndDiscard();
144 }
145
146 }
147
148 }