Replaced set/get*Listener on Channel interface with add/remove*Listener;

added 'collection' of listeners of three types: Return, Flow and Confirm;
single private monitor manages these collections;
copy of collection is made under lock before making listener calls.

Author: Steve Powell

src/com/rabbitmq/client/Channel.java

@@ -111,40 +111,46 @@ public interface Channel extends ShutdownNotifier {
     void abort(int closeCode, String closeMessage) throws IOException;
 
     /**
-     * Return the current {@link ReturnListener}.
-     * @return an interface to the current return listener
+     * Add a {@link ReturnListener}.
+     * @param listener the listener to add
      */
-    ReturnListener getReturnListener();
+    void addReturnListener(ReturnListener listener);
 
     /**
-     * Set the current {@link ReturnListener}.
-     * @param listener the listener to use, or null indicating "don't use one".
+     * Remove a {@link ReturnListener}.
+     * @param listener the listener to remove
+     * @return <code><b>true</b></code> if the listener was found and removed,
+     * <code><b>false</b></code> otherwise
      */
-    void setReturnListener(ReturnListener listener);
+    boolean removeReturnListener(ReturnListener listener);
 
     /**
-     * Return the current {@link FlowListener}.
-     * @return an interface to the current flow listener.
+     * Add a {@link FlowListener}.
+     * @param listener the listener to add
      */
-    FlowListener getFlowListener();
+    void addFlowListener(FlowListener listener);
 
     /**
-     * Set the current {@link FlowListener}.
-     * @param listener the listener to use, or null indicating "don't use one".
+     * Remove a {@link FlowListener}.
+     * @param listener the listener to remove
+     * @return <code><b>true</b></code> if the listener was found and removed,
+     * <code><b>false</b></code> otherwise
      */
-    void setFlowListener(FlowListener listener);
+    boolean removeFlowListener(FlowListener listener);
 
     /**
-     * Return the current {@link ConfirmListener}.
-     * @return an interface to the current ack listener.
+     * Add a {@link ConfirmListener}.
+     * @param listener the listener to add
      */
-    ConfirmListener getConfirmListener();
+    void addConfirmListener(ConfirmListener listener);
 
     /**
-     * Set the current {@link ConfirmListener}.
-     * @param listener the listener to use, or null indicating "don't use one".
+     * Remove a {@link ConfirmListener}.
+     * @param listener the listener to remove
+     * @return <code><b>true</b></code> if the listener was found and removed,
+     * <code><b>false</b></code> otherwise
      */
-    void setConfirmListener(ConfirmListener listener);
+    boolean removeConfirmListener(ConfirmListener listener);
 
     /**
      * Get the current default consumer. @see setDefaultConsumer for rationale.

src/com/rabbitmq/client/impl/ChannelN.java

@@ -18,8 +18,11 @@
 package com.rabbitmq.client.impl;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
@@ -78,25 +81,19 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
     public final Map<String, Consumer> _consumers =
         Collections.synchronizedMap(new HashMap<String, Consumer>());
 
-    /** Reference to the currently-active ReturnListener, or null if there is none.
-     */
-    public volatile ReturnListener returnListener = null;
-
-    /** Reference to the currently-active FlowListener, or null if there is none.
-     */
-    public volatile FlowListener flowListener = null;
-
-    /** Reference to the currently-active ConfirmListener, or null if there is none.
-     */
-    public volatile ConfirmListener confirmListener = null;
+    /** Monitor protecting all listeners collections */
+    private final Object listenersMonitor = new Object();
+        /** The ReturnListener collection. */
+        private final Collection<ReturnListener> returnListeners = new LinkedList<ReturnListener>();
+        /** The FlowListener collection. */
+        private final Collection<FlowListener> flowListeners = new LinkedList<FlowListener>();
+        /** The ConfirmListener collection. */
+        private final Collection<ConfirmListener> confirmListeners = new LinkedList<ConfirmListener>();
 
-    /** Sequence number of next published message requiring confirmation.
-     */
+    /** Sequence number of next published message requiring confirmation. */
     private long nextPublishSeqNo = 0L;
 
-    /** Reference to the currently-active default consumer, or null if there is
-     *  none.
-     */
+    /** The current default consumer, or null if there is none. */
     public volatile Consumer defaultConsumer = null;
 
     /**
@@ -123,43 +120,67 @@ public void open() throws IOException {
         Utility.use(openOk);
     }
 
-    /** Returns the current ReturnListener. */
-    public ReturnListener getReturnListener() {
-        return returnListener;
+    /** Returns a readable copy of the current ReturnListener collection. */
+    private Collection<ReturnListener> getReturnListeners() {
+        synchronized (listenersMonitor) {
+            if (returnListeners.isEmpty())
+                return Collections.emptyList();
+            return new ArrayList<ReturnListener>(returnListeners);
+        }
     }
 
-    /**
-     * Sets the current ReturnListener.
-     * A null argument is interpreted to mean "do not use a return listener".
-     */
-    public void setReturnListener(ReturnListener listener) {
-        returnListener = listener;
+    public void addReturnListener(ReturnListener listener) {
+        synchronized (listenersMonitor) {
+            returnListeners.add(listener);
+        }
     }
 
-    /** Returns the current {@link FlowListener}. */
-    public FlowListener getFlowListener() {
-        return flowListener;
+    public boolean removeReturnListener(ReturnListener listener) {
+        synchronized (listenersMonitor) {
+            return returnListeners.remove(listener);
+        }
     }
 
-    /**
-     * Sets the current {@link FlowListener}.
-     * A null argument is interpreted to mean "do not use a flow listener".
-     */
-    public void setFlowListener(FlowListener listener) {
-        flowListener = listener;
+    /** Returns a readable copy of the current FlowListener collection. */
+    private Collection<FlowListener> getFlowListeners() {
+        synchronized (listenersMonitor) {
+            if (flowListeners.isEmpty())
+                return Collections.emptyList();
+            return new ArrayList<FlowListener>(flowListeners);
+        }
     }
 
-    /** Returns the current {@link ConfirmListener}. */
-    public ConfirmListener getConfirmListener() {
-        return confirmListener;
+    public void addFlowListener(FlowListener listener) {
+        synchronized (listenersMonitor) {
+            flowListeners.add(listener);
+        }
     }
 
-    /**
-     * Sets the current {@link ConfirmListener}.
-     * A null argument is interpreted to mean "do not use a confirm listener".
-     */
-    public void setConfirmListener(ConfirmListener listener) {
-        confirmListener = listener;
+    public boolean removeFlowListener(FlowListener listener) {
+        synchronized (listenersMonitor) {
+            return flowListeners.remove(listener);
+        }
+    }
+
+    /** Returns a readable copy of the current ConfirmListener collection. */
+    private Collection<ConfirmListener> getConfirmListeners() {
+        synchronized (listenersMonitor) {
+            if (confirmListeners.isEmpty())
+                return Collections.emptyList();
+            return new ArrayList<ConfirmListener>(confirmListeners);
+        }
+    }
+
+    public void addConfirmListener(ConfirmListener listener) {
+        synchronized (listenersMonitor) {
+            confirmListeners.add(listener);
+        }
+    }
+
+    public boolean removeConfirmListener(ConfirmListener listener) {
+        synchronized (listenersMonitor) {
+            return confirmListeners.remove(listener);
+        }
     }
 
     /** Returns the current default consumer. */
@@ -275,22 +296,7 @@ public void releaseChannelNumber() {
                 }
                 return true;
             } else if (method instanceof Basic.Return) {
-                ReturnListener l = getReturnListener();
-                if (l != null) {
-                    Basic.Return basicReturn = (Basic.Return) method;
-                    try {
-                        l.handleReturn(basicReturn.getReplyCode(),
-                                            basicReturn.getReplyText(),
-                                            basicReturn.getExchange(),
-                                            basicReturn.getRoutingKey(),
-                                            (BasicProperties)
-                                            command.getContentHeader(),
-                                            command.getContentBody());
-                    } catch (Throwable ex) {
-                        _connection.getExceptionHandler().handleReturnListenerException(this,
-                                                                                        ex);
-                    }
-                }
+                callReturnListeners(command, (Basic.Return) method);
                 return true;
             } else if (method instanceof Channel.Flow) {
                 Channel.Flow channelFlow = (Channel.Flow) method;
@@ -299,36 +305,13 @@ public void releaseChannelNumber() {
                     transmit(new Channel.FlowOk(!_blockContent));
                     _channelMutex.notifyAll();
                 }
-                FlowListener l = getFlowListener();
-                if (l != null) {
-                    try {
-                        l.handleFlow(channelFlow.getActive());
-                    } catch (Throwable ex) {
-                        _connection.getExceptionHandler().handleFlowListenerException(this, ex);
-                    }
-                }
+                callFlowListeners(command, channelFlow);
                 return true;
             } else if (method instanceof Basic.Ack) {
-                Basic.Ack ack = (Basic.Ack) method;
-                ConfirmListener l = getConfirmListener();
-                if (l != null) {
-                    try {
-                        l.handleAck(ack.getDeliveryTag(), ack.getMultiple());
-                    } catch (Throwable ex) {
-                        _connection.getExceptionHandler().handleConfirmListenerException(this, ex);
-                    }
-                }
+                callConfirmListeners(command, (Basic.Ack) method);
                 return true;
             } else if (method instanceof Basic.Nack) {
-                Basic.Nack nack = (Basic.Nack) method;
-                ConfirmListener l = getConfirmListener();
-                if (l != null) {
-                    try {
-                        l.handleNack(nack.getDeliveryTag(), nack.getMultiple());
-                    } catch (Throwable ex) {
-                        _connection.getExceptionHandler().handleConfirmListenerException(this, ex);
-                    }
-                }
+                callConfirmListeners(command, (Basic.Nack) method);
                 return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
@@ -379,6 +362,51 @@ public void releaseChannelNumber() {
         }
     }
 
+    private void callReturnListeners(Command command, Basic.Return basicReturn) {
+        try {
+            for (ReturnListener l : getReturnListeners()) {
+                l.handleReturn(basicReturn.getReplyCode(),
+                               basicReturn.getReplyText(),
+                               basicReturn.getExchange(),
+                               basicReturn.getRoutingKey(),
+             (BasicProperties) command.getContentHeader(),
+                               command.getContentBody());
+            }
+        } catch (Throwable ex) {
+            _connection.getExceptionHandler().handleReturnListenerException(this, ex);
+        }
+    }
+
+    private void callFlowListeners(Command command, Channel.Flow channelFlow) {
+        try {
+            for (FlowListener l : getFlowListeners()) {
+                l.handleFlow(channelFlow.getActive());
+            }
+        } catch (Throwable ex) {
+            _connection.getExceptionHandler().handleFlowListenerException(this, ex);
+        }
+    }
+
+    private void callConfirmListeners(Command command, Basic.Ack ack) {
+        try {
+            for (ConfirmListener l : getConfirmListeners()) {
+                l.handleAck(ack.getDeliveryTag(), ack.getMultiple());
+            }
+        } catch (Throwable ex) {
+            _connection.getExceptionHandler().handleConfirmListenerException(this, ex);
+        }
+    }
+
+    private void callConfirmListeners(Command command, Basic.Nack nack) {
+        try {
+            for (ConfirmListener l : getConfirmListeners()) {
+                l.handleNack(nack.getDeliveryTag(), nack.getMultiple());
+            }
+        } catch (Throwable ex) {
+            _connection.getExceptionHandler().handleConfirmListenerException(this, ex);
+        }
+    }
+
     private void asyncShutdown(Command command) throws IOException {
         releaseChannelNumber();
         ShutdownSignalException signal = new ShutdownSignalException(false,

test/src/com/rabbitmq/client/test/ConfirmBase.java

@@ -32,7 +32,7 @@ public class ConfirmBase extends BrokerTestCase {
     protected void setUp() throws IOException {
         super.setUp();
         unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
-        channel.setConfirmListener(new ConfirmListener() {
+        channel.addConfirmListener(new ConfirmListener() {
                 public void handleAck(long seqNo, boolean multiple) {
                     if (!unconfirmedSet.contains(seqNo)) {
                         fail("got duplicate ack: " + seqNo);

test/src/com/rabbitmq/client/test/functional/AlternateExchange.java

@@ -57,7 +57,7 @@ private static boolean[] expected(String key) {
 
     @Override protected void setUp() throws IOException {
         super.setUp();
-        channel.setReturnListener(new ReturnListener() {
+        channel.addReturnListener(new ReturnListener() {
                 public void handleReturn(int replyCode,
                                          String replyText,
                                          String exchange,

test/src/com/rabbitmq/client/test/performance/ScalabilityTest.java

@@ -284,7 +284,7 @@ private float timeRouting(Channel channel, String[] routingKeys)
         boolean mandatory = true;
         boolean immdediate = true;
         final CountDownLatch latch = new CountDownLatch(params.messageCount);
-        channel.setReturnListener(new ReturnListener() {
+        channel.addReturnListener(new ReturnListener() {
                 public void handleReturn(int replyCode, String replyText,
                                          String exchange, String routingKey,
                                          AMQP.BasicProperties properties, byte[] body) throws IOException {

test/src/com/rabbitmq/examples/ConfirmDontLoseMessages.java

@@ -61,7 +61,7 @@ public void run() {
                 Connection conn = connectionFactory.newConnection();
                 Channel ch = conn.createChannel();
                 ch.queueDeclare(QUEUE_NAME, true, false, false, null);
-                ch.setConfirmListener(new ConfirmListener() {
+                ch.addConfirmListener(new ConfirmListener() {
                         public void handleAck(long seqNo, boolean multiple) {
                             if (multiple) {
                                 unconfirmedSet.headSet(seqNo+1).clear();

test/src/com/rabbitmq/examples/MulticastMain.java

@@ -135,8 +135,8 @@ public static void main(String[] args) {
                                                 1000L * samplingInterval,
                                                 rateLimit, minMsgSize, timeLimit,
                                                 confirm);
-                channel.setReturnListener(p);
-                channel.setConfirmListener(p);
+                channel.addReturnListener(p);
+                channel.addConfirmListener(p);
                 Thread t = new Thread(p);
                 producerThreads[i] = t;
                 t.start();

test/src/com/rabbitmq/examples/TestMain.java

@@ -230,7 +230,7 @@ public void run() throws IOException {
 
         _ch1 = createChannel();
 
-        _ch1.setReturnListener(new ReturnListener() {
+        _ch1.addReturnListener(new ReturnListener() {
             public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body)
                     throws IOException {
                 Method method = new AMQImpl.Basic.Return(replyCode, replyText, exchange, routingKey);