@Test public void invokeCallableService() throws Exception { ThreadAnalyzer analyzer = new ThreadAnalyzer(); @SuppressWarnings("unchecked") Function<Executor, String> service = Mockito.mock(Function.class); strategy.invokeService(executor, e -> { analyzer.checkServiceInvocation(); return service.apply(e); }).toFuture().get(); analyzer.verifyNoErrors(); }
@Override public Publisher<Object> invokeService( final Executor fallback, StreamingHttpRequest request, final Function<StreamingHttpRequest, Publisher<Object>> service, final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_RECEIVE_DATA)) { request = request.transformRawPayloadBody(payload -> payload.publishOn(e)); } Publisher<Object> resp; if (offloaded(OFFLOAD_RECEIVE_META)) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.apply(r).subscribeShareContext()) .recoverWith(cause -> succeeded(errorHandler.apply(cause, e))) .flatMapPublisher(identity()); } else { resp = service.apply(request); } if (offloaded(OFFLOAD_SEND)) { resp = resp.subscribeOn(e); } return resp; }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public Publisher<Object> invokeService( final Executor fallback, StreamingHttpRequest request, final Function<StreamingHttpRequest, Publisher<Object>> service, final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_RECEIVE_DATA)) { request = request.transformRawPayloadBody(payload -> payload.publishOn(e)); } Publisher<Object> resp; if (offloaded(OFFLOAD_RECEIVE_META)) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.apply(r).subscribeShareContext()) .recoverWith(cause -> succeeded(errorHandler.apply(cause, e))) .flatMapPublisher(identity()); } else { resp = service.apply(request); } if (offloaded(OFFLOAD_SEND)) { resp = resp.subscribeOn(e); } return resp; } }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public Publisher<Object> invokeService( final Executor fallback, StreamingHttpRequest request, final Function<StreamingHttpRequest, Publisher<Object>> service, final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_RECEIVE_DATA)) { request = request.transformRawPayloadBody(payload -> payload.publishOn(e)); } Publisher<Object> resp; if (offloaded(OFFLOAD_RECEIVE_META)) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.apply(r).subscribeShareContext()) .recoverWith(cause -> succeeded(errorHandler.apply(cause, e))) .flatMapPublisher(identity()); } else { resp = service.apply(request); } if (offloaded(OFFLOAD_SEND)) { resp = resp.subscribeOn(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 Publisher<Object> invokeService( final Executor fallback, StreamingHttpRequest request, final Function<StreamingHttpRequest, Publisher<Object>> service, final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_RECEIVE_DATA)) { request = request.transformRawPayloadBody(payload -> payload.publishOn(e)); } Publisher<Object> resp; if (offloaded(OFFLOAD_RECEIVE_META)) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.apply(r).subscribeShareContext()) .recoverWith(cause -> succeeded(errorHandler.apply(cause, e))) .flatMapPublisher(identity()); } else { resp = service.apply(request); } if (offloaded(OFFLOAD_SEND)) { resp = resp.subscribeOn(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 Publisher<Object> invokeService( final Executor fallback, StreamingHttpRequest request, final Function<StreamingHttpRequest, Publisher<Object>> service, final BiFunction<Throwable, Executor, Publisher<Object>> errorHandler) { final Executor e = executor(fallback); if (offloaded(OFFLOAD_RECEIVE_DATA)) { request = request.transformRawPayloadBody(payload -> payload.publishOn(e)); } Publisher<Object> resp; if (offloaded(OFFLOAD_RECEIVE_META)) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.apply(r).subscribeShareContext()) .recoverWith(cause -> succeeded(errorHandler.apply(cause, e))) .flatMapPublisher(identity()); } else { resp = service.apply(request); } if (offloaded(OFFLOAD_SEND)) { resp = resp.subscribeOn(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();  }