class BlockingStreamingToStreamingService extends AbstractServiceAdapterHolder {

    @Override
    public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx,
                                                final StreamingHttpRequest request,
                                                final StreamingHttpResponseFactory responseFactory) {

        return new Single<StreamingHttpResponse>() {
            @Override
            protected void handleSubscribe(final Subscriber<? super StreamingHttpResponse> subscriber) {
                final ThreadInterruptingCancellable tiCancellable = new ThreadInterruptingCancellable(currentThread());
                try {
                    subscriber.onSubscribe(tiCancellable);
                } catch (Throwable cause) {
                    handleExceptionFromOnSubscribe(subscriber, cause);
                    return;
                }

                final Processor payloadProcessor = newCompletableProcessor();
                DefaultBlockingStreamingHttpServerResponse response = null;
                BufferHttpPayloadWriter payloadWriterOuter = null;
                try {
                    final BufferHttpPayloadWriter payloadWriter = payloadWriterOuter = new BufferHttpPayloadWriter(
                            ctx.headersFactory().newTrailers(), payloadProcessor);

                    final Consumer<HttpResponseMetaData> sendMeta = (metaData) -> {
                        final StreamingHttpResponse result;
                        try {
                            // transfer-encoding takes precedence over content-length.
                            // > When a message does not have a Transfer-Encoding header field, a
                            // Content-Length header field can provide the anticipated size.
                            // https://tools.ietf.org/html/rfc7230#section-3.3.2
                            HttpHeaders headers = metaData.headers();
                            boolean addTrailers = isTransferEncodingChunked(headers);
                            if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) &&
                                    // HEAD responses MUST never carry a payload, adding chunked makes no sense and
                                    // breaks our HttpResponseDecoder
                                    // TODO(jayv) we can provide an optimized PayloadWriter for HEAD response that
                                    // immediately completes on sendMeta() and disallows further writes or trailers
                                    !HEAD.equals(request.method())) {
                                // this is likely not supported in http/1.0 and it is possible that a response has
                                // neither header and the connection close indicates the end of the response.
                                // https://tools.ietf.org/html/rfc7230#section-3.3.3
                                headers.add(TRANSFER_ENCODING, CHUNKED);
                                addTrailers = true;
                            }
                            Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect()
                                    .map(buffer -> (Object) buffer)); // down cast to Object
                            if (addTrailers) {
                                payload = payload.concat(succeeded(payloadWriter.trailers()));
                            }
                            payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() {
                                @Override
                                public void request(final long n) {
                                    // Noop
                                }

                                @Override
                                public void cancel() {
                                    tiCancellable.cancel();
                                }
                            });
                            result = newTransportResponse(metaData.status(), metaData.version(),
                                    metaData.headers(), ctx.executionContext().bufferAllocator(),
                                    payload, ctx.headersFactory());
                        } catch (Throwable t) {
                            subscriber.onError(t);
                            throw t;
                        }
                        subscriber.onSuccess(result);
                    };

                    response = new DefaultBlockingStreamingHttpServerResponse(OK, request.version(),
                                    ctx.headersFactory().newHeaders(), payloadWriter,
                                    ctx.executionContext().bufferAllocator(), sendMeta);
                    original.handle(ctx, request.toBlockingStreamingRequest(), response);
                } catch (Throwable cause) {
                    tiCancellable.setDone(cause);
                    if (response == null || response.markMetaSent()) {
                        safeOnError(subscriber, cause);
                    } else if (payloadWriterOuter.markSubscriberComplete()) {
                        payloadProcessor.onError(cause);
                    } else {
                        LOGGER.error("An exception occurred after the response was sent", cause);
                    }
                    return;
                }
                tiCancellable.setDone();
            }
        };
    }

      BlockingStreamingToStreamingService(final BlockingStreamingHttpService original,
                                        final HttpExecutionStrategyInfluencer influencer);

}