(NETTY-12) Only set the responder state to be sent when it’s going to write to the output channel.

- Also call BodyProducer.onError if exception raised when calling BodyProducer.getContentLength

Author: chtyim

src/main/java/co/cask/http/BasicHttpResponder.java

@@ -67,7 +67,6 @@ public BasicHttpResponder(Channel channel, boolean keepAlive) {
 
   @Override
   public ChunkResponder sendChunkStart(HttpResponseStatus status, @Nullable Multimap<String, String> headers) {
-    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     Preconditions.checkArgument((status.getCode() >= 200 && status.getCode() < 210) , "Http Chunk Failure");
     HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
 
@@ -79,14 +78,14 @@ public ChunkResponder sendChunkStart(HttpResponseStatus status, @Nullable Multim
     }
 
     boolean responseKeepAlive = setResponseKeepAlive(response);
+    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     channel.write(response);
     return new ChannelChunkResponder(channel, responseKeepAlive);
   }
 
   @Override
   public void sendContent(HttpResponseStatus status, @Nullable ChannelBuffer content, String contentType,
                           @Nullable Multimap<String, String> headers) {
-    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
 
     setCustomHeaders(response, headers);
@@ -100,6 +99,7 @@ public void sendContent(HttpResponseStatus status, @Nullable ChannelBuffer conte
     }
 
     boolean responseKeepAlive = setResponseKeepAlive(response);
+    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     ChannelFuture future = channel.write(response);
     if (!responseKeepAlive) {
       future.addListener(ChannelFutureListener.CLOSE);
@@ -108,14 +108,15 @@ public void sendContent(HttpResponseStatus status, @Nullable ChannelBuffer conte
 
   @Override
   public void sendFile(File file, @Nullable Multimap<String, String> headers) {
-    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
 
     setCustomHeaders(response, headers);
     response.setHeader(HttpHeaders.Names.CONTENT_LENGTH, file.length());
 
     final boolean responseKeepAlive = setResponseKeepAlive(response);
 
+    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
+
     // Write the initial line and the header.
     channel.write(response);
 
@@ -144,11 +145,11 @@ public void operationComplete(ChannelFuture future) throws Exception {
   @Override
   public void sendContent(HttpResponseStatus status, final BodyProducer bodyProducer,
                           @Nullable Multimap<String, String> headers) {
-    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
     final long contentLength;
     try {
       contentLength = bodyProducer.getContentLength();
     } catch (Throwable t) {
+      bodyProducer.handleError(t);
       // Response with error and close the connection
       sendContent(HttpResponseStatus.INTERNAL_SERVER_ERROR,
                   ChannelBuffers.wrappedBuffer(
@@ -173,6 +174,9 @@ public void sendContent(HttpResponseStatus status, final BodyProducer bodyProduc
     boolean responseKeepAlive = setResponseKeepAlive(response);
     final ChannelFutureListener completionListener = createBodyProducerCompletionListener(bodyProducer,
                                                                                           responseKeepAlive);
+
+    Preconditions.checkArgument(responded.compareAndSet(false, true), "Response has been already sent");
+
     // Streams the data produced by the given BodyProducer
     channel.write(response).addListener(new ChannelFutureListener() {
 

src/main/java/co/cask/http/RequestRouter.java

@@ -134,9 +134,7 @@ private boolean handleRequest(HttpRequest httpRequest, Channel channel, ChannelH
   public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
     String exceptionMessage = "Exception caught in channel processing.";
     Throwable cause = e.getCause();
-    if (!exceptionRaised.get()) {
-      exceptionRaised.set(true);
-
+    if (exceptionRaised.compareAndSet(false, true)) {
       if (methodInfo != null) {
         LOG.error(exceptionMessage, cause);
         methodInfo.sendError(HttpResponseStatus.INTERNAL_SERVER_ERROR, cause);