Mailing-List: contact commits-help@camel.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@camel.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r1420612 - in /camel/branches/camel-2.10.x: ./
 components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/
Date: Wed, 12 Dec 2012 10:38:37 -0000
To: commits@camel.apache.org
From: davsclaus@apache.org
Message-Id: <20121212103838.3A43E238896F@eris.apache.org>

Author: davsclaus
Date: Wed Dec 12 10:38:36 2012
New Revision: 1420612

URL: http://svn.apache.org/viewvc?rev=1420612&view=rev
Log:
Make it easier to customize and reuse ServerChannelHandler in camel-netty.

Added:
    camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
      - copied unchanged from r1420610, camel/trunk/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
Modified:
    camel/branches/camel-2.10.x/   (props changed)
    camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java

Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
  Merged /camel/trunk:r1420610

Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.

Modified: camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java?rev=1420612&r1=1420611&r2=1420612&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java (original)
+++ camel/branches/camel-2.10.x/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java Wed Dec 12 10:38:36 2012
@@ -19,16 +19,13 @@ package org.apache.camel.component.netty
 import java.net.SocketAddress;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelExchangeException;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
-import org.apache.camel.component.netty.NettyConstants;
 import org.apache.camel.component.netty.NettyConsumer;
 import org.apache.camel.component.netty.NettyHelper;
 import org.apache.camel.component.netty.NettyPayloadHelper;
 import org.apache.camel.util.CamelLogger;
 import org.apache.camel.util.IOHelper;
-import org.jboss.netty.channel.ChannelFuture;
 import org.jboss.netty.channel.ChannelFutureListener;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelStateEvent;
@@ -66,6 +63,7 @@ public class ServerChannelHandler extend
         if (LOG.isTraceEnabled()) {
             LOG.trace("Channel closed: {}", e.getChannel());
         }
+        // to keep track of open sockets
         consumer.getAllChannels().remove(e.getChannel());
     }
 
@@ -167,7 +165,7 @@ public class ServerChannelHandler extend
             }
 
             // we got a body to write
-            ChannelFutureListener listener = new ResponseFutureListener(exchange, messageEvent.getRemoteAddress());
+            ChannelFutureListener listener = createResponseFutureListener(consumer, exchange, messageEvent.getRemoteAddress());
             if (consumer.getConfiguration().isTcp()) {
                 NettyHelper.writeBodyAsync(LOG, messageEvent.getChannel(), null, body, exchange, listener);
             } else {
@@ -177,47 +175,15 @@ public class ServerChannelHandler extend
     }
 
     /**
-     * A {@link ChannelFutureListener} that performs the disconnect logic when
-     * sending the response is complete.
+     * Creates the {@link ChannelFutureListener} to execute when writing the response is complete.
+     *
+     * @param consumer          the netty consumer
+     * @param exchange          the exchange
+     * @param remoteAddress     the remote address of the message
+     * @return the listener.
      */
-    private final class ResponseFutureListener implements ChannelFutureListener {
-
-        private final Exchange exchange;
-        private final SocketAddress remoteAddress;
-
-        private ResponseFutureListener(Exchange exchange, SocketAddress remoteAddress) {
-            this.exchange = exchange;
-            this.remoteAddress = remoteAddress;
-        }
-
-        @Override
-        public void operationComplete(ChannelFuture future) throws Exception {
-            // if it was not a success then thrown an exception
-            if (!future.isSuccess()) {
-                Exception e = new CamelExchangeException("Cannot write response to " + remoteAddress, exchange, future.getCause());
-                consumer.getExceptionHandler().handleException(e);
-            }
-
-            // should channel be closed after complete?
-            Boolean close;
-            if (exchange.hasOut()) {
-                close = exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
-            } else {
-                close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
-            }
-
-            // should we disconnect, the header can override the configuration
-            boolean disconnect = consumer.getConfiguration().isDisconnect();
-            if (close != null) {
-                disconnect = close;
-            }
-            if (disconnect) {
-                if (LOG.isTraceEnabled()) {
-                    LOG.trace("Closing channel when complete at address: {}", remoteAddress);
-                }
-                NettyHelper.close(future.getChannel());
-            }
-        }
+    protected ChannelFutureListener createResponseFutureListener(NettyConsumer consumer, Exchange exchange, SocketAddress remoteAddress) {
+        return new ServerResponseFutureListener(consumer, exchange, remoteAddress);
     }
 
 }