@Test public void wrapServiceThatWasNotOffloaded() throws Exception { ThreadAnalyzer analyzer = new ThreadAnalyzer(); StreamingHttpService svc = strategy.offloadService(executor, (ctx, request, responseFactory) -> { analyzer.checkContext(ctx); analyzer.checkServiceInvocation(); return succeeded(analyzer.createNewResponse() .payloadBody(analyzer.instrumentedRequestPayloadForServer(request.payloadBody()))); }); StreamingHttpRequest req = analyzer.createNewRequest(); DefaultStreamingHttpRequestResponseFactory respFactory = new DefaultStreamingHttpRequestResponseFactory(DEFAULT_ALLOCATOR, INSTANCE, HTTP_1_1); TestHttpServiceContext ctx = new TestHttpServiceContext(INSTANCE, respFactory, new ExecutionContextToHttpExecutionContext(contextRule, noOffloadsStrategy())); analyzer.instrumentedResponseForServer(svc.handle(ctx, req, ctx.streamingResponseFactory())) .flatMapPublisher(StreamingHttpResponse::payloadBody) .toFuture().get(); analyzer.verify(); }
@Override public StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service) { return new StreamingHttpService() { private final Executor e = executor(fallback); @Override public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx, StreamingHttpRequest request, final StreamingHttpResponseFactory responseFactory) { final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(), DefaultHttpExecutionStrategy.this); final HttpServiceContext wrappedCtx = new ExecutionContextOverridingServiceContext(ctx, DefaultHttpExecutionStrategy.this, e); if (diff == null) { return service.handle(wrappedCtx, request, responseFactory); } else { if (diff.isDataReceiveOffloaded()) { request = request.transformRawPayloadBody(p -> p.publishOn(e)); } final Single<StreamingHttpResponse> resp; if (diff.isMetadataReceiveOffloaded()) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.handle(wrappedCtx, r, responseFactory).subscribeShareContext()) .flatMap(identity()); } else { resp = service.handle(wrappedCtx, request, responseFactory); } return diff.isSendOffloaded() ? resp.map(r -> r.transformRawPayloadBody(p -> p.subscribeOn(e))).subscribeOn(e) : resp; } } @Override public Completable closeAsync() { return service.closeAsync(); } @Override public Completable closeAsyncGracefully() { return service.closeAsyncGracefully(); } }; }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service) { return new StreamingHttpService() { private final Executor e = executor(fallback); @Override public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx, StreamingHttpRequest request, final StreamingHttpResponseFactory responseFactory) { final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(), DefaultHttpExecutionStrategy.this); final HttpServiceContext wrappedCtx = new ExecutionContextOverridingServiceContext(ctx, DefaultHttpExecutionStrategy.this, e); if (diff == null) { return service.handle(wrappedCtx, request, responseFactory); } else { if (diff.isDataReceiveOffloaded()) { request = request.transformRawPayloadBody(p -> p.publishOn(e)); } final Single<StreamingHttpResponse> resp; if (diff.isMetadataReceiveOffloaded()) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.handle(wrappedCtx, r, responseFactory).subscribeShareContext()) .flatMap(identity()); } else { resp = service.handle(wrappedCtx, request, responseFactory); } return diff.isSendOffloaded() ? resp.map(r -> r.transformRawPayloadBody(p -> p.subscribeOn(e))).subscribeOn(e) : resp; } } @Override public Completable closeAsync() { return service.closeAsync(); } @Override public Completable closeAsyncGracefully() { return service.closeAsyncGracefully(); } }; } }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service) { return new StreamingHttpService() { private final Executor e = executor(fallback); @Override public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx, StreamingHttpRequest request, final StreamingHttpResponseFactory responseFactory) { final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(), DefaultHttpExecutionStrategy.this); final HttpServiceContext wrappedCtx = new ExecutionContextOverridingServiceContext(ctx, DefaultHttpExecutionStrategy.this, e); if (diff == null) { return service.handle(wrappedCtx, request, responseFactory); } else { if (diff.isDataReceiveOffloaded()) { request = request.transformRawPayloadBody(p -> p.publishOn(e)); } final Single<StreamingHttpResponse> resp; if (diff.isMetadataReceiveOffloaded()) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.handle(wrappedCtx, r, responseFactory).subscribeShareContext()) .flatMap(identity()); } else { resp = service.handle(wrappedCtx, request, responseFactory); } return diff.isSendOffloaded() ? resp.map(r -> r.transformRawPayloadBody(p -> p.subscribeOn(e))).subscribeOn(e) : resp; } } @Override public Completable closeAsync() { return service.closeAsync(); } @Override public Completable closeAsyncGracefully() { return service.closeAsyncGracefully(); } }; } DefaultHttpExecutionStrategy(@Nullable final Executor executor, final byte offloads, final boolean threadAffinity,
                                 final MergeStrategy mergeStrategy);   DefaultHttpExecutionStrategy(byte offloadOverride, HttpExecutionStrategy original); }
DefaultHttpExecutionStrategy implements HttpExecutionStrategy { @Override public StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service) { return new StreamingHttpService() { private final Executor e = executor(fallback); @Override public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx, StreamingHttpRequest request, final StreamingHttpResponseFactory responseFactory) { final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(), DefaultHttpExecutionStrategy.this); final HttpServiceContext wrappedCtx = new ExecutionContextOverridingServiceContext(ctx, DefaultHttpExecutionStrategy.this, e); if (diff == null) { return service.handle(wrappedCtx, request, responseFactory); } else { if (diff.isDataReceiveOffloaded()) { request = request.transformRawPayloadBody(p -> p.publishOn(e)); } final Single<StreamingHttpResponse> resp; if (diff.isMetadataReceiveOffloaded()) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.handle(wrappedCtx, r, responseFactory).subscribeShareContext()) .flatMap(identity()); } else { resp = service.handle(wrappedCtx, request, responseFactory); } return diff.isSendOffloaded() ? resp.map(r -> r.transformRawPayloadBody(p -> p.subscribeOn(e))).subscribeOn(e) : resp; } } @Override public Completable closeAsync() { return service.closeAsync(); } @Override public Completable closeAsyncGracefully() { return service.closeAsyncGracefully(); } }; } 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 StreamingHttpService offloadService(final Executor fallback, final StreamingHttpService service) { return new StreamingHttpService() { private final Executor e = executor(fallback); @Override public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx, StreamingHttpRequest request, final StreamingHttpResponseFactory responseFactory) { final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(), DefaultHttpExecutionStrategy.this); final HttpServiceContext wrappedCtx = new ExecutionContextOverridingServiceContext(ctx, DefaultHttpExecutionStrategy.this, e); if (diff == null) { return service.handle(wrappedCtx, request, responseFactory); } else { if (diff.isDataReceiveOffloaded()) { request = request.transformRawPayloadBody(p -> p.publishOn(e)); } final Single<StreamingHttpResponse> resp; if (diff.isMetadataReceiveOffloaded()) { final StreamingHttpRequest r = request; resp = e.submit(() -> service.handle(wrappedCtx, r, responseFactory).subscribeShareContext()) .flatMap(identity()); } else { resp = service.handle(wrappedCtx, request, responseFactory); } return diff.isSendOffloaded() ? resp.map(r -> r.transformRawPayloadBody(p -> p.subscribeOn(e))).subscribeOn(e) : resp; } } @Override public Completable closeAsync() { return service.closeAsync(); } @Override public Completable closeAsyncGracefully() { return service.closeAsyncGracefully(); } }; } 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();  }