@Test public void throwAfterSendMetaData() throws Exception { CountDownLatch onErrorLatch = new CountDownLatch(1); AtomicReference<Throwable> throwableRef = new AtomicReference<>(); BlockingStreamingHttpService syncService = (ctx, request, response) -> { response.sendMetaData(); throw DELIBERATE_EXCEPTION; }; StreamingHttpService asyncService = toStreamingHttpService(syncService, strategy -> strategy).adaptor(); StreamingHttpResponse asyncResponse = asyncService.handle(mockCtx, reqRespFactory.get("/"), reqRespFactory) .subscribeOn(executorRule.executor()).toFuture().get(); assertMetaData(OK, asyncResponse); toSource(asyncResponse.payloadBody()).subscribe(new Subscriber<Buffer>() { @Override public void onSubscribe(final Subscription s) { } @Override public void onNext(final Buffer s) { } @Override public void onError(final Throwable t) { throwableRef.set(t); onErrorLatch.countDown(); } @Override public void onComplete() { } }); onErrorLatch.await(); assertThat(throwableRef.get(), is(DELIBERATE_EXCEPTION)); }
@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 { HttpHeaders headers = metaData.headers(); boolean addTrailers = isTransferEncodingChunked(headers); if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) && !HEAD.equals(request.method())) { headers.add(TRANSFER_ENCODING, CHUNKED); addTrailers = true; } Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect() .map(buffer -> (Object) buffer)); if (addTrailers) { payload = payload.concat(succeeded(payloadWriter.trailers())); } payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() { @Override public void request(final long n) { } @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 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 { HttpHeaders headers = metaData.headers(); boolean addTrailers = isTransferEncodingChunked(headers); if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) && !HEAD.equals(request.method())) { headers.add(TRANSFER_ENCODING, CHUNKED); addTrailers = true; } Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect() .map(buffer -> (Object) buffer)); if (addTrailers) { payload = payload.concat(succeeded(payloadWriter.trailers())); } payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() { @Override public void request(final long n) { } @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 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 { HttpHeaders headers = metaData.headers(); boolean addTrailers = isTransferEncodingChunked(headers); if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) && !HEAD.equals(request.method())) { headers.add(TRANSFER_ENCODING, CHUNKED); addTrailers = true; } Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect() .map(buffer -> (Object) buffer)); if (addTrailers) { payload = payload.concat(succeeded(payloadWriter.trailers())); } payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() { @Override public void request(final long n) { } @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); }
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 { HttpHeaders headers = metaData.headers(); boolean addTrailers = isTransferEncodingChunked(headers); if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) && !HEAD.equals(request.method())) { headers.add(TRANSFER_ENCODING, CHUNKED); addTrailers = true; } Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect() .map(buffer -> (Object) buffer)); if (addTrailers) { payload = payload.concat(succeeded(payloadWriter.trailers())); } payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() { @Override public void request(final long n) { } @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); @Override Single<StreamingHttpResponse> handle(final HttpServiceContext ctx,
                                                final StreamingHttpRequest request,
                                                final StreamingHttpResponseFactory responseFactory); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); }
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 { HttpHeaders headers = metaData.headers(); boolean addTrailers = isTransferEncodingChunked(headers); if (!addTrailers && !HTTP_1_0.equals(metaData.version()) && !hasContentLength(headers) && !HEAD.equals(request.method())) { headers.add(TRANSFER_ENCODING, CHUNKED); addTrailers = true; } Publisher<Object> payload = fromSource(payloadProcessor).merge(payloadWriter.connect() .map(buffer -> (Object) buffer)); if (addTrailers) { payload = payload.concat(succeeded(payloadWriter.trailers())); } payload = payload.beforeSubscription(() -> new PublisherSource.Subscription() { @Override public void request(final long n) { } @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); @Override Single<StreamingHttpResponse> handle(final HttpServiceContext ctx,
                                                final StreamingHttpRequest request,
                                                final StreamingHttpResponseFactory responseFactory); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully();  }