class 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) {
                // We compute the difference between the ExecutionStrategy from the current ExecutionContext and this
                // ExecutionStrategy to understand if we need to offload more then we already offloaded:
                final HttpExecutionStrategy diff = difference(fallback, ctx.executionContext().executionStrategy(),
                        DefaultHttpExecutionStrategy.this);
                // The service should see this ExecutionStrategy inside the ExecutionContext:
                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())
                                // exec.submit() returns a Single<Single<response>>, so flatten the nested Single.
                                .flatMap(identity());
                    } else {
                        resp = service.handle(wrappedCtx, request, responseFactory);
                    }
                    return diff.isSendOffloaded() ?
                            // This is different as compared to invokeService() where we just offload once on the
                            // flattened (meta + data) stream. In this case, we need to preserve the service contract
                            // and hence have to offload both meta and data separately.
                            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);

}