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 static org.hamcrest.MatcherAssert.assertThat;
30
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.MinimalHttpAsyncClient;
38 import org.apache.hc.client5.http.protocol.HttpClientContext;
39 import org.apache.hc.core5.http.ContentType;
40 import org.apache.hc.core5.http.HttpHost;
41 import org.apache.hc.core5.http.HttpResponse;
42 import org.apache.hc.core5.http.Message;
43 import org.apache.hc.core5.http.Method;
44 import org.apache.hc.core5.http.URIScheme;
45 import org.apache.hc.core5.http.config.Http1Config;
46 import org.apache.hc.core5.http.nio.AsyncClientEndpoint;
47 import org.apache.hc.core5.http.nio.entity.AsyncEntityProducers;
48 import org.apache.hc.core5.http.nio.entity.BasicAsyncEntityConsumer;
49 import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
50 import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
51 import org.apache.hc.core5.http2.config.H2Config;
52 import org.apache.hc.core5.reactive.ReactiveServerExchangeHandler;
53 import org.apache.hc.core5.testing.nio.H2TestServer;
54 import org.apache.hc.core5.testing.reactive.ReactiveEchoProcessor;
55 import org.hamcrest.CoreMatchers;
56 import org.junit.jupiter.api.Test;
57
58 public abstract class TestHttp1ReactiveMinimal extends AbstractHttpReactiveFundamentalsTest<MinimalHttpAsyncClient> {
59
60 public TestHttp1ReactiveMinimal(final URIScheme scheme) {
61 super(scheme);
62 }
63
64 @Override
65 protected H2TestServer startServer() throws Exception {
66 return startServer(Http1Config.DEFAULT, null, null);
67 }
68
69 @Override
70 protected MinimalHttpAsyncClient startClient() throws Exception {
71 return startMinimalClient(
72 Http1Config.DEFAULT,
73 H2Config.DEFAULT,
74 b -> {});
75 }
76
77 @Test
78 public void testConcurrentPostRequestsSameEndpoint() throws Exception {
79 final H2TestServer server = startServer();
80 server.register("/echo/*", () ->
81 new ReactiveServerExchangeHandler(new ReactiveEchoProcessor()));
82 final HttpHost target = targetHost();
83
84 final MinimalHttpAsyncClient client = startClient();
85
86 final byte[] b1 = new byte[1024];
87 final Random rnd = new Random(System.currentTimeMillis());
88 rnd.nextBytes(b1);
89
90 final int reqCount = 20;
91
92 final Future<AsyncClientEndpoint> endpointLease = client.lease(target, null);
93 final AsyncClientEndpoint endpoint = endpointLease.get(5, TimeUnit.SECONDS);
94 try {
95 final Queue<Future<Message<HttpResponse, byte[]>>> queue = new LinkedList<>();
96 for (int i = 0; i < reqCount; i++) {
97 final Future<Message<HttpResponse, byte[]>> future = endpoint.execute(
98 new BasicRequestProducer(Method.GET, target, "/echo/",
99 AsyncEntityProducers.create(b1, ContentType.APPLICATION_OCTET_STREAM)),
100 new BasicResponseConsumer<>(new BasicAsyncEntityConsumer()), HttpClientContext.create(), null);
101 queue.add(future);
102 }
103 while (!queue.isEmpty()) {
104 final Future<Message<HttpResponse, byte[]>> future = queue.remove();
105 final Message<HttpResponse, byte[]> responseMessage = future.get();
106 assertThat(responseMessage, CoreMatchers.notNullValue());
107 final HttpResponse response = responseMessage.getHead();
108 assertThat(response.getCode(), CoreMatchers.equalTo(200));
109 final byte[] b2 = responseMessage.getBody();
110 assertThat(b1, CoreMatchers.equalTo(b2));
111 endpoint.releaseAndReuse();
112 }
113 } finally {
114 endpoint.releaseAndDiscard();
115 }
116
117 }
118
119 }