ConcurrencyLimitingClient extends AbstractConcurrencyLimitingClient<HttpRequest, HttpResponse> implements HttpClient { public static Function<? super HttpClient, ConcurrencyLimitingClient> newDecorator(int maxConcurrency) { validateMaxConcurrency(maxConcurrency); return delegate -> new ConcurrencyLimitingClient(delegate, maxConcurrency); } ConcurrencyLimitingClient(HttpClient delegate, int maxConcurrency); private  ConcurrencyLimitingClient(HttpClient delegate,
                                      int maxConcurrency, long timeout, TimeUnit unit); static Function<? super HttpClient, ConcurrencyLimitingClient> newDecorator(int maxConcurrency); static Function<? super HttpClient, ConcurrencyLimitingClient> newDecorator(
            int maxConcurrency, long timeout, TimeUnit unit);  }
@Test void testOrdinaryRequest() throws Exception { final HttpRequest req = newReq(); final ClientRequestContext ctx = newContext(); final HttpResponseWriter actualRes = HttpResponse.streaming(); when(delegate.execute(ctx, req)).thenReturn(actualRes); final ConcurrencyLimitingClient client = newDecorator(1).apply(delegate); assertThat(client.numActiveRequests()).isZero(); final HttpResponse res = client.execute(ctx, req); assertThat(res.isOpen()).isTrue(); assertThat(client.numActiveRequests()).isEqualTo(1); closeAndDrain(actualRes, res); assertThat(res.isOpen()).isFalse(); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }
@Test void testLimitedRequest() throws Exception { final ClientRequestContext ctx1 = newContext(); final ClientRequestContext ctx2 = newContext(); final HttpRequest req1 = newReq(); final HttpRequest req2 = newReq(); final HttpResponseWriter actualRes1 = HttpResponse.streaming(); final HttpResponseWriter actualRes2 = HttpResponse.streaming(); when(delegate.execute(ctx1, req1)).thenReturn(actualRes1); when(delegate.execute(ctx2, req2)).thenReturn(actualRes2); final ConcurrencyLimitingClient client = newDecorator(1).apply(delegate); final HttpResponse res1 = client.execute(ctx1, req1); verify(delegate).execute(ctx1, req1); assertThat(res1.isOpen()).isTrue(); final HttpResponse res2 = client.execute(ctx2, req2); verify(delegate, never()).execute(ctx2, req2); assertThat(res2.isOpen()).isTrue(); assertThat(client.numActiveRequests()).isEqualTo(1); closeAndDrain(actualRes1, res1); await().untilAsserted(() -> verify(delegate).execute(ctx2, req2)); assertThat(client.numActiveRequests()).isEqualTo(1); closeAndDrain(actualRes2, res2); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }
@Test void testTimeout() throws Exception { final ClientRequestContext ctx1 = newContext(); final ClientRequestContext ctx2 = newContext(); final HttpRequest req1 = newReq(); final HttpRequest req2 = newReq(); final HttpResponseWriter actualRes1 = HttpResponse.streaming(); final HttpResponseWriter actualRes2 = HttpResponse.streaming(); when(delegate.execute(ctx1, req1)).thenReturn(actualRes1); final ConcurrencyLimitingClient client = newDecorator(1, 500, TimeUnit.MILLISECONDS).apply(delegate); final HttpResponse res1 = client.execute(ctx1, req1); final HttpResponse res2 = client.execute(ctx2, req2); Thread.sleep(1000); res2.subscribe(NoopSubscriber.get()); assertThatThrownBy(() -> res2.whenComplete().join()) .hasCauseInstanceOf(UnprocessedRequestException.class) .hasRootCauseInstanceOf(RequestTimeoutException.class); assertThat(res2.isOpen()).isFalse(); res1.subscribe(NoopSubscriber.get()); assertThat(res1.isOpen()).isTrue(); assertThat(res1.whenComplete()).isNotDone(); actualRes1.close(); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }
@Test void testFaultyDelegate() throws Exception { final ClientRequestContext ctx = newContext(); final HttpRequest req = newReq(); when(delegate.execute(ctx, req)).thenThrow(Exception.class); final ConcurrencyLimitingClient client = newDecorator(1).apply(delegate); assertThat(client.numActiveRequests()).isZero(); final HttpResponse res = client.execute(ctx, req); res.subscribe(NoopSubscriber.get()); assertThat(res.isOpen()).isFalse(); assertThatThrownBy(() -> res.whenComplete().get()).hasCauseInstanceOf(Exception.class); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }
@Test void testUnlimitedRequest() throws Exception { final ClientRequestContext ctx = newContext(); final HttpRequest req = newReq(); final HttpResponseWriter actualRes = HttpResponse.streaming(); when(delegate.execute(ctx, req)).thenReturn(actualRes); final ConcurrencyLimitingClient client = newDecorator(0).apply(delegate); final HttpResponse res = client.execute(ctx, req); verify(delegate).execute(ctx, req); assertThat(res.isOpen()).isTrue(); assertThat(client.numActiveRequests()).isEqualTo(1); closeAndDrain(actualRes, res); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }
@Test void testUnlimitedRequestWithFaultyDelegate() throws Exception { final ClientRequestContext ctx = newContext(); final HttpRequest req = newReq(); when(delegate.execute(ctx, req)).thenThrow(Exception.class); final ConcurrencyLimitingClient client = newDecorator(0).apply(delegate); assertThatThrownBy(() -> client.execute(ctx, req)).isInstanceOf(Exception.class); verify(delegate).execute(ctx, req); await().untilAsserted(() -> assertThat(client.numActiveRequests()).isZero()); }