@Test public void invokeClient() throws Exception { ThreadAnalyzer analyzer = new ThreadAnalyzer(); StreamingHttpRequest req = analyzer.createNewRequest(); StreamingHttpResponse resp = analyzer.createNewResponse(); analyzer.instrumentedResponseForClient(strategy.invokeClient(executor, from(req, req.payloadBodyAndTrailers()), null, (publisher, __) -> analyzer.instrumentedFlatRequestForClient(publisher).ignoreElements().concat(succeeded(resp)))) .flatMapPublisher(StreamingHttpResponse::payloadBody) .toFuture().get(); analyzer.verify(); }
@Override public <FS> Single<StreamingHttpResponse> invokeClient( final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy, final ClientInvoker<FS> client) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_SEND)) { flattenedRequest = flattenedRequest.subscribeOn(e); } Single<StreamingHttpResponse> resp = client.invokeClient(flattenedRequest, flushStrategy); if (offloaded(OFFLOAD_RECEIVE_META)) { resp = resp.publishOn(e); } if (offloaded(OFFLOAD_RECEIVE_DATA)) { resp = resp.map(response -> response.transformRawPayloadBody(payload -> payload.publishOn(e))); } return resp; }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public <FS> Single<StreamingHttpResponse> invokeClient( final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy, final ClientInvoker<FS> client) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_SEND)) { flattenedRequest = flattenedRequest.subscribeOn(e); } Single<StreamingHttpResponse> resp = client.invokeClient(flattenedRequest, flushStrategy); if (offloaded(OFFLOAD_RECEIVE_META)) { resp = resp.publishOn(e); } if (offloaded(OFFLOAD_RECEIVE_DATA)) { resp = resp.map(response -> response.transformRawPayloadBody(payload -> payload.publishOn(e))); } return resp; } }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public <FS> Single<StreamingHttpResponse> invokeClient( final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy, final ClientInvoker<FS> client) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_SEND)) { flattenedRequest = flattenedRequest.subscribeOn(e); } Single<StreamingHttpResponse> resp = client.invokeClient(flattenedRequest, flushStrategy); if (offloaded(OFFLOAD_RECEIVE_META)) { resp = resp.publishOn(e); } if (offloaded(OFFLOAD_RECEIVE_DATA)) { resp = resp.map(response -> response.transformRawPayloadBody(payload -> payload.publishOn(e))); } return resp; } DefaultHttpExecutionStrategy(@Nullable final Executor executor, final byte offloads, final boolean threadAffinity,
                                 final MergeStrategy mergeStrategy);   DefaultHttpExecutionStrategy(byte offloadOverride, HttpExecutionStrategy original); }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public <FS> Single<StreamingHttpResponse> invokeClient( final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy, final ClientInvoker<FS> client) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_SEND)) { flattenedRequest = flattenedRequest.subscribeOn(e); } Single<StreamingHttpResponse> resp = client.invokeClient(flattenedRequest, flushStrategy); if (offloaded(OFFLOAD_RECEIVE_META)) { resp = resp.publishOn(e); } if (offloaded(OFFLOAD_RECEIVE_DATA)) { resp = resp.map(response -> response.transformRawPayloadBody(payload -> payload.publishOn(e))); } return resp; } DefaultHttpExecutionStrategy(@Nullable final Executor executor, final byte offloads, final boolean threadAffinity,
                                 final MergeStrategy mergeStrategy);   DefaultHttpExecutionStrategy(byte offloadOverride, HttpExecutionStrategy original); @Nullable @Override Executor executor(); @Override Single<StreamingHttpResponse> invokeClient(
            final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy,
            final ClientInvoker<FS> client); @Override Publisher<Object> invokeService(
            final Executor fallback, StreamingHttpRequest request,
            final Function<StreamingHttpRequest, Publisher<Object>> service,
            final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler); @Override StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service); @Override boolean isMetadataReceiveOffloaded(); @Override boolean isDataReceiveOffloaded(); @Override boolean isSendOffloaded(); @Override HttpExecutionStrategy merge(final HttpExecutionStrategy other); @Override Single<T> invokeService(final Executor fallback, final Function<Executor, T> service); @Override Single<T> offloadSend(final Executor fallback, final Single<T> original); @Override Single<T> offloadReceive(final Executor fallback, final Single<T> original); @Override Publisher<T> offloadSend(final Executor fallback, final Publisher<T> original); @Override Publisher<T> offloadReceive(final Executor fallback, final Publisher<T> original); @Override boolean equals(final Object o); @Override int hashCode(); @Override String toString(); }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public <FS> Single<StreamingHttpResponse> invokeClient( final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy, final ClientInvoker<FS> client) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_SEND)) { flattenedRequest = flattenedRequest.subscribeOn(e); } Single<StreamingHttpResponse> resp = client.invokeClient(flattenedRequest, flushStrategy); if (offloaded(OFFLOAD_RECEIVE_META)) { resp = resp.publishOn(e); } if (offloaded(OFFLOAD_RECEIVE_DATA)) { resp = resp.map(response -> response.transformRawPayloadBody(payload -> payload.publishOn(e))); } return resp; } DefaultHttpExecutionStrategy(@Nullable final Executor executor, final byte offloads, final boolean threadAffinity,
                                 final MergeStrategy mergeStrategy);   DefaultHttpExecutionStrategy(byte offloadOverride, HttpExecutionStrategy original); @Nullable @Override Executor executor(); @Override Single<StreamingHttpResponse> invokeClient(
            final Executor fallback, Publisher<Object> flattenedRequest, @Nullable final FS flushStrategy,
            final ClientInvoker<FS> client); @Override Publisher<Object> invokeService(
            final Executor fallback, StreamingHttpRequest request,
            final Function<StreamingHttpRequest, Publisher<Object>> service,
            final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler); @Override StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service); @Override boolean isMetadataReceiveOffloaded(); @Override boolean isDataReceiveOffloaded(); @Override boolean isSendOffloaded(); @Override HttpExecutionStrategy merge(final HttpExecutionStrategy other); @Override Single<T> invokeService(final Executor fallback, final Function<Executor, T> service); @Override Single<T> offloadSend(final Executor fallback, final Single<T> original); @Override Single<T> offloadReceive(final Executor fallback, final Single<T> original); @Override Publisher<T> offloadSend(final Executor fallback, final Publisher<T> original); @Override Publisher<T> offloadReceive(final Executor fallback, final Publisher<T> original); @Override boolean equals(final Object o); @Override int hashCode(); @Override String toString();  }