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.core5.testing.reactive;
28
29 import static java.lang.String.format;
30 import static org.hamcrest.MatcherAssert.assertThat;
31
32 import java.io.ByteArrayOutputStream;
33 import java.io.IOException;
34 import java.net.InetSocketAddress;
35 import java.net.SocketTimeoutException;
36 import java.net.URI;
37 import java.nio.ByteBuffer;
38 import java.nio.channels.Channels;
39 import java.nio.channels.WritableByteChannel;
40 import java.util.List;
41 import java.util.Random;
42 import java.util.concurrent.CancellationException;
43 import java.util.concurrent.ExecutionException;
44 import java.util.concurrent.Future;
45 import java.util.concurrent.atomic.AtomicBoolean;
46 import java.util.concurrent.atomic.AtomicReference;
47
48 import io.reactivex.rxjava3.core.Flowable;
49 import io.reactivex.rxjava3.core.Observable;
50 import org.apache.hc.core5.function.Supplier;
51 import org.apache.hc.core5.http.HttpResponse;
52 import org.apache.hc.core5.http.HttpStreamResetException;
53 import org.apache.hc.core5.http.Message;
54 import org.apache.hc.core5.http.Method;
55 import org.apache.hc.core5.http.URIScheme;
56 import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
57 import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncServer;
58 import org.apache.hc.core5.http.impl.routing.RequestRouter;
59 import org.apache.hc.core5.http.nio.AsyncServerExchangeHandler;
60 import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
61 import org.apache.hc.core5.http2.HttpVersionPolicy;
62 import org.apache.hc.core5.reactive.ReactiveEntityProducer;
63 import org.apache.hc.core5.reactive.ReactiveResponseConsumer;
64 import org.apache.hc.core5.reactive.ReactiveServerExchangeHandler;
65 import org.apache.hc.core5.reactor.IOReactorConfig;
66 import org.apache.hc.core5.reactor.ListenerEndpoint;
67 import org.apache.hc.core5.testing.nio.extension.H2AsyncRequesterResource;
68 import org.apache.hc.core5.testing.nio.extension.H2AsyncServerResource;
69 import org.apache.hc.core5.testing.reactive.Reactive3TestUtils.StreamDescription;
70 import org.apache.hc.core5.util.TextUtils;
71 import org.apache.hc.core5.util.Timeout;
72 import org.hamcrest.CoreMatchers;
73 import org.junit.jupiter.api.Assertions;
74 import org.junit.jupiter.api.Test;
75 import org.junit.jupiter.api.extension.RegisterExtension;
76 import org.reactivestreams.Publisher;
77
78 public abstract class ReactiveClientTest {
79
80 private static final Timeout SOCKET_TIMEOUT = Timeout.ofSeconds(30);
81 private static final Timeout RESULT_TIMEOUT = Timeout.ofSeconds(60);
82
83 private static final Random RANDOM = new Random();
84
85 private final HttpVersionPolicy versionPolicy;
86 @RegisterExtension
87 private final H2AsyncServerResource serverResource;
88 @RegisterExtension
89 private final H2AsyncRequesterResource clientResource;
90
91 public ReactiveClientTest(final HttpVersionPolicy httpVersionPolicy) {
92 this.versionPolicy = httpVersionPolicy;
93 this.serverResource = new H2AsyncServerResource(bootstrap -> bootstrap
94 .setVersionPolicy(versionPolicy)
95 .setIOReactorConfig(
96 IOReactorConfig.custom()
97 .setSoTimeout(SOCKET_TIMEOUT)
98 .build())
99 .setRequestRouter(RequestRouter.<Supplier<AsyncServerExchangeHandler>>builder()
100 .addRoute(RequestRouter.LOCAL_AUTHORITY, "*", () -> new ReactiveServerExchangeHandler(new ReactiveEchoProcessor()))
101 .resolveAuthority(RequestRouter.LOCAL_AUTHORITY_RESOLVER)
102 .build())
103 );
104 this.clientResource = new H2AsyncRequesterResource(bootstrap -> bootstrap
105 .setVersionPolicy(versionPolicy)
106 .setIOReactorConfig(IOReactorConfig.custom()
107 .setSoTimeout(SOCKET_TIMEOUT)
108 .build())
109 );
110 }
111
112 @Test
113 public void testSimpleRequest() throws Exception {
114 final InetSocketAddress address = startServer();
115 final HttpAsyncRequester requester = clientResource.start();
116 final byte[] input = new byte[1024];
117 RANDOM.nextBytes(input);
118 final Publisher<ByteBuffer> publisher = Flowable.just(ByteBuffer.wrap(input));
119 final ReactiveEntityProducer producer = new ReactiveEntityProducer(publisher, input.length, null, null);
120
121 final BasicRequestProducer request = getRequestProducer(address, producer);
122
123 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
124 requester.execute(request, consumer, SOCKET_TIMEOUT, null);
125
126 final Message<HttpResponse, Publisher<ByteBuffer>> response = consumer.getResponseFuture()
127 .get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit());
128
129 final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
130 final WritableByteChannel writableByteChannel = Channels.newChannel(byteArrayOutputStream);
131 for (final ByteBuffer byteBuffer : Observable.fromPublisher(response.getBody()).toList().blockingGet()) {
132 writableByteChannel.write(byteBuffer);
133 }
134 writableByteChannel.close();
135 final byte[] output = byteArrayOutputStream.toByteArray();
136 Assertions.assertArrayEquals(input, output);
137 }
138
139 private BasicRequestProducer getRequestProducer(final InetSocketAddress address, final ReactiveEntityProducer producer) {
140 return new BasicRequestProducer(Method.POST, URI.create("http://localhost:" + address.getPort()), producer);
141 }
142
143 @Test
144 public void testLongRunningRequest() throws Exception {
145 final InetSocketAddress address = startServer();
146 final HttpAsyncRequester requester = clientResource.start();
147 final long expectedLength = 6_554_200L;
148 final AtomicReference<String> expectedHash = new AtomicReference<>();
149 final Flowable<ByteBuffer> stream = Reactive3TestUtils.produceStream(expectedLength, expectedHash);
150 final ReactiveEntityProducer producer = new ReactiveEntityProducer(stream, -1, null, null);
151 final BasicRequestProducer request = getRequestProducer(address, producer);
152
153 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
154 requester.execute(request, consumer, SOCKET_TIMEOUT, null);
155 final Message<HttpResponse, Publisher<ByteBuffer>> response = consumer.getResponseFuture()
156 .get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit());
157 final StreamDescription desc = Reactive3TestUtils.consumeStream(response.getBody()).blockingGet();
158
159 Assertions.assertEquals(expectedLength, desc.length);
160 Assertions.assertEquals(expectedHash.get(), TextUtils.toHexString(desc.md.digest()));
161 }
162
163 @Test
164 public void testManySmallBuffers() throws Exception {
165
166
167
168
169 final InetSocketAddress address = startServer();
170 final HttpAsyncRequester requester = clientResource.start();
171 for (int i = 0; i < 10; i++) {
172 final long expectedLength = 1_024_000;
173 final int maximumBlockSize = 1024;
174 final AtomicReference<String> expectedHash = new AtomicReference<>();
175 final Publisher<ByteBuffer> stream = Reactive3TestUtils.produceStream(expectedLength, maximumBlockSize, expectedHash);
176 final ReactiveEntityProducer producer = new ReactiveEntityProducer(stream, -1, null, null);
177 final BasicRequestProducer request = getRequestProducer(address, producer);
178
179 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
180 requester.execute(request, consumer, SOCKET_TIMEOUT, null);
181 final Message<HttpResponse, Publisher<ByteBuffer>> response = consumer.getResponseFuture()
182 .get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit());
183 final StreamDescription desc = Reactive3TestUtils.consumeStream(response.getBody()).blockingGet();
184
185 Assertions.assertEquals(expectedLength, desc.length);
186 Assertions.assertEquals(expectedHash.get(), TextUtils.toHexString(desc.md.digest()));
187 }
188 }
189
190 @Test
191 public void testRequestError() throws Exception {
192 final InetSocketAddress address = startServer();
193 final HttpAsyncRequester requester = clientResource.start();
194 final RuntimeException exceptionThrown = new RuntimeException("Test");
195 final Publisher<ByteBuffer> publisher = Flowable.error(exceptionThrown);
196 final ReactiveEntityProducer producer = new ReactiveEntityProducer(publisher, 100, null, null);
197
198 final BasicRequestProducer request = getRequestProducer(address, producer);
199
200 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
201
202 final Future<Void> future = requester.execute(request, consumer, SOCKET_TIMEOUT, null);
203
204 final ExecutionException exception = Assertions.assertThrows(ExecutionException.class, () ->
205 future.get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
206 Assertions.assertTrue(exception.getCause() instanceof HttpStreamResetException);
207 Assertions.assertSame(exceptionThrown, exception.getCause().getCause());
208 }
209
210 @Test
211 public void testRequestTimeout() throws Exception {
212 final InetSocketAddress address = startServer();
213 final HttpAsyncRequester requester = clientResource.start();
214 final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean(false);
215 final Publisher<ByteBuffer> publisher = Flowable.<ByteBuffer>never()
216 .doOnCancel(() -> requestPublisherWasCancelled.set(true));
217 final ReactiveEntityProducer producer = new ReactiveEntityProducer(publisher, -1, null, null);
218 final BasicRequestProducer request = getRequestProducer(address, producer);
219
220 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
221 final Future<Void> future = requester.execute(request, consumer, Timeout.ofSeconds(1), null);
222
223 final ExecutionException exception = Assertions.assertThrows(ExecutionException.class, () ->
224 future.get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
225 Assertions.assertTrue(requestPublisherWasCancelled.get());
226 final Throwable cause = exception.getCause();
227 if (versionPolicy == HttpVersionPolicy.FORCE_HTTP_1) {
228 Assertions.assertTrue(cause instanceof SocketTimeoutException, "Expected SocketTimeoutException, but got " + cause.getClass().getName());
229 } else if (versionPolicy == HttpVersionPolicy.FORCE_HTTP_2) {
230 Assertions.assertTrue(cause instanceof HttpStreamResetException, format("Expected RST_STREAM, but %s was thrown", cause.getClass().getName()));
231 } else {
232 Assertions.fail("Unknown HttpVersionPolicy: " + versionPolicy);
233 }
234 }
235
236 @Test
237 public void testResponseCancellation() throws Exception {
238 final InetSocketAddress address = startServer();
239 final HttpAsyncRequester requester = clientResource.start();
240 final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean(false);
241 final AtomicReference<Throwable> requestStreamError = new AtomicReference<>();
242 final Publisher<ByteBuffer> stream = Reactive3TestUtils.produceStream(Long.MAX_VALUE, 1024, null)
243 .doOnCancel(() -> requestPublisherWasCancelled.set(true))
244 .doOnError(requestStreamError::set);
245 final ReactiveEntityProducer producer = new ReactiveEntityProducer(stream, -1, null, null);
246 final BasicRequestProducer request = getRequestProducer(address, producer);
247
248 final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
249 final Future<Void> future = requester.execute(request, consumer, SOCKET_TIMEOUT, null);
250 final Message<HttpResponse, Publisher<ByteBuffer>> response = consumer.getResponseFuture()
251 .get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit());
252
253 final AtomicBoolean responsePublisherWasCancelled = new AtomicBoolean(false);
254 final List<ByteBuffer> outputBuffers = Flowable.fromPublisher(response.getBody())
255 .doOnCancel(() -> responsePublisherWasCancelled.set(true))
256 .take(3)
257 .toList()
258 .blockingGet();
259 Assertions.assertEquals(3, outputBuffers.size());
260 Assertions.assertTrue(responsePublisherWasCancelled.get(), "The response subscription should have been cancelled");
261 final Exception exception = Assertions.assertThrows(Exception.class, () ->
262 future.get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
263 assertThat(exception, CoreMatchers.anyOf(
264 CoreMatchers.instanceOf(CancellationException.class),
265 CoreMatchers.instanceOf(ExecutionException.class)));
266 Assertions.assertTrue(exception.getCause() instanceof HttpStreamResetException);
267 Assertions.assertTrue(requestPublisherWasCancelled.get());
268 Assertions.assertNull(requestStreamError.get());
269 }
270
271 private InetSocketAddress startServer() throws IOException, InterruptedException, ExecutionException {
272 final HttpAsyncServer server = serverResource.start();
273 final ListenerEndpoint listener = server.listen(new InetSocketAddress(0), URIScheme.HTTP).get();
274 final InetSocketAddress address = (InetSocketAddress) listener.getAddress();
275 return address;
276 }
277 }