Allow to send response from request consumer

HttpAsyncService.patch

diff --git a/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java b/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java
index 8d6a23a..d85aa0b 100644
--- a/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java
+++ b/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java
@@ -98,6 +98,7 @@ import org.apache.http.util.Asserts;
  */
 @Immutable // provided injected dependencies are immutable
 public class HttpAsyncService implements NHttpServerEventHandler {
+    public static final String HTTP_EXCHANGE = "http.nio.http-exchange";
 
     static final String HTTP_EXCHANGE_STATE = "http.nio.http-exchange-state";
 
@@ -278,6 +279,11 @@ public class HttpAsyncService implements NHttpServerEventHandler {
         this.httpProcessor.process(request, context);
 
         final HttpAsyncRequestHandler<Object> requestHandler = getRequestHandler(request);
+        final HttpResponse response = this.responseFactory.newHttpResponse(HttpVersion.HTTP_1_1,
+                HttpStatus.SC_OK, context);
+        final HttpAsyncExchangeImpl httpExchange = new HttpAsyncExchangeImpl(
+                request, response, state, conn, context);
+        context.setAttribute(HTTP_EXCHANGE, httpExchange);
         final HttpAsyncRequestConsumer<Object> consumer = requestHandler.processRequest(request, context);
         consumer.requestReceived(request);
 
@@ -375,43 +381,48 @@ public class HttpAsyncService implements NHttpServerEventHandler {
             if (state.getResponseState() == MessageState.READY) {
                 final Queue<PipelineEntry> pipeline = state.getPipeline();
                 final PipelineEntry pipelineEntry = pipeline.poll();
-                if (pipelineEntry == null) {
+                if (pipelineEntry == null && state.getOutgoing() == null) {
                     conn.suspendOutput();
                     return;
                 }
                 state.setResponseState(MessageState.INIT);
-                final Object result = pipelineEntry.getResult();
-                final HttpRequest request = pipelineEntry.getRequest();
-                final HttpContext context = pipelineEntry.getContext();
-                if (result != null) {
-                    final HttpResponse response = this.responseFactory.newHttpResponse(HttpVersion.HTTP_1_1,
-                            HttpStatus.SC_OK, context);
-                    final HttpAsyncExchangeImpl httpExchange = new HttpAsyncExchangeImpl(
-                            request, response, state, conn, context);
-                    final HttpAsyncRequestHandler<Object> handler = pipelineEntry.getHandler();
-                    conn.suspendOutput();
-                    try {
-                        handler.handle(result, httpExchange, context);
-                    } catch (RuntimeException ex) {
-                        throw ex;
-                    } catch (Exception ex) {
-                        pipeline.add(new PipelineEntry(
-                            request,
-                            null,
-                            ex,
-                            handler,
-                            context));
-                        state.setResponseState(MessageState.READY);
-                        responseReady(conn);
-                        return;
+                if (pipelineEntry != null) {
+                    final Object result = pipelineEntry.getResult();
+                    final HttpRequest request = pipelineEntry.getRequest();
+                    final HttpContext context = pipelineEntry.getContext();
+                    if (result != null) {
+                        HttpAsyncExchange httpExchange = (HttpAsyncExchange) context.getAttribute(HTTP_EXCHANGE);
+                        if (httpExchange == null) {
+                            final HttpResponse response = this.responseFactory.newHttpResponse(HttpVersion.HTTP_1_1,
+                                    HttpStatus.SC_OK, context);
+                            httpExchange = new HttpAsyncExchangeImpl(
+                                    request, response, state, conn, context);
+                        }
+                        final HttpAsyncRequestHandler<Object> handler = pipelineEntry.getHandler();
+                        conn.suspendOutput();
+                        try {
+                            handler.handle(result, httpExchange, context);
+                        } catch (RuntimeException ex) {
+                            throw ex;
+                        } catch (Exception ex) {
+                            pipeline.add(new PipelineEntry(
+                                request,
+                                null,
+                                ex,
+                                handler,
+                                context));
+                            state.setResponseState(MessageState.READY);
+                            responseReady(conn);
+                            return;
+                        }
+                    } else {
+                        final Exception exception = pipelineEntry.getException();
+                        final HttpAsyncResponseProducer responseProducer = handleException(
+                                exception != null ? exception : new HttpException("Internal error processing request"),
+                                context);
+                        final HttpResponse error = responseProducer.generateResponse();
+                        state.setOutgoing(new Outgoing(request, error, responseProducer, context));
                     }
-                } else {
-                    final Exception exception = pipelineEntry.getException();
-                    final HttpAsyncResponseProducer responseProducer = handleException(
-                            exception != null ? exception : new HttpException("Internal error processing request"),
-                            context);
-                    final HttpResponse error = responseProducer.generateResponse();
-                    state.setOutgoing(new Outgoing(request, error, responseProducer, context));
                 }
             }
             if (state.getResponseState() == MessageState.INIT) {

NHttpReverseProxy.patch

--- httpcore-nio/src/examples/org/apache/http/examples/nio/NHttpReverseProxy.java	2014-12-07 22:43:43.667158705 +0300
+++ /tmp/NHttpReverseProxy.java	2014-12-07 22:44:59.320706033 +0300
@@ -360,6 +360,8 @@
             }
             synchronized (httpExchange) {
                 httpExchange.reset();
+                HttpAsyncExchange responseTrigger = (HttpAsyncExchange) context.getAttribute(HttpAsyncService.HTTP_EXCHANGE);
+                httpExchange.setResponseTrigger(responseTrigger);
                 String id = String.format("%08X", this.counter.getAndIncrement());
                 httpExchange.setId(id);
                 httpExchange.setTarget(this.target);