Merge in default

Author: Steve Powell

src/com/rabbitmq/client/Channel.java

@@ -111,40 +111,61 @@ 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.
+     * Remove all {@link ReturnListener}s.
      */
-    FlowListener getFlowListener();
+    void clearReturnListeners();
 
     /**
-     * Set the current {@link FlowListener}.
-     * @param listener the listener to use, or null indicating "don't use one".
+     * Add a {@link FlowListener}.
+     * @param listener the listener to add
      */
-    void setFlowListener(FlowListener listener);
+    void addFlowListener(FlowListener listener);
 
     /**
-     * Return the current {@link ConfirmListener}.
-     * @return an interface to the current ack listener.
+     * 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
      */
-    ConfirmListener getConfirmListener();
+    boolean removeFlowListener(FlowListener listener);
 
     /**
-     * Set the current {@link ConfirmListener}.
-     * @param listener the listener to use, or null indicating "don't use one".
+     * Remove all {@link FlowListener}s.
      */
-    void setConfirmListener(ConfirmListener listener);
+    void clearFlowListeners();
+
+    /**
+     * Add a {@link ConfirmListener}.
+     * @param listener the listener to add
+     */
+    void addConfirmListener(ConfirmListener listener);
+
+    /**
+     * 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
+     */
+    boolean removeConfirmListener(ConfirmListener listener);
+
+    /**
+     * Remove all {@link ConfirmListener}s.
+     */
+    void clearConfirmListeners();
 
     /**
      * Get the current default consumer. @see setDefaultConsumer for rationale.

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

@@ -18,9 +18,11 @@
 package com.rabbitmq.client.impl;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP;
@@ -78,25 +80,18 @@ 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;
+    /* All listeners collections are in CopyOnWriteArrayList objects */
+    /** The ReturnListener collection. */
+    private final Collection<ReturnListener> returnListeners = new CopyOnWriteArrayList<ReturnListener>();
+    /** The FlowListener collection. */
+    private final Collection<FlowListener> flowListeners = new CopyOnWriteArrayList<FlowListener>();
+    /** The ConfirmListener collection. */
+    private final Collection<ConfirmListener> confirmListeners = new CopyOnWriteArrayList<ConfirmListener>();
 
-    /** Reference to the currently-active ConfirmListener, or null if there is none.
-     */
-    public volatile ConfirmListener confirmListener = null;
-
-    /** 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 +118,40 @@ public void open() throws IOException {
         Utility.use(openOk);
     }
 
-    /** Returns the current ReturnListener. */
-    public ReturnListener getReturnListener() {
-        return returnListener;
+    public void addReturnListener(ReturnListener listener) {
+        returnListeners.add(listener);
     }
 
-    /**
-     * 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 boolean removeReturnListener(ReturnListener listener) {
+        return returnListeners.remove(listener);
     }
 
-    /** Returns the current {@link FlowListener}. */
-    public FlowListener getFlowListener() {
-        return flowListener;
+    public void clearReturnListeners() {
+        returnListeners.clear();
     }
 
-    /**
-     * 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;
+    public void addFlowListener(FlowListener listener) {
+        flowListeners.add(listener);
     }
 
-    /** Returns the current {@link ConfirmListener}. */
-    public ConfirmListener getConfirmListener() {
-        return confirmListener;
+    public boolean removeFlowListener(FlowListener listener) {
+        return flowListeners.remove(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 void clearFlowListeners() {
+        flowListeners.clear();
+    }
+
+    public void addConfirmListener(ConfirmListener listener) {
+        confirmListeners.add(listener);
+    }
+
+    public boolean removeConfirmListener(ConfirmListener listener) {
+        return confirmListeners.remove(listener);
+    }
+
+    public void clearConfirmListeners() {
+        confirmListeners.clear();
     }
 
     /** Returns the current default consumer. */
@@ -275,22 +267,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 +276,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 +333,51 @@ public void releaseChannelNumber() {
         }
     }
 
+    private void callReturnListeners(Command command, Basic.Return basicReturn) {
+        try {
+            for (ReturnListener l : this.returnListeners) {
+                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 : this.flowListeners) {
+                l.handleFlow(channelFlow.getActive());
+            }
+        } catch (Throwable ex) {
+            _connection.getExceptionHandler().handleFlowListenerException(this, ex);
+        }
+    }
+
+    private void callConfirmListeners(Command command, Basic.Ack ack) {
+        try {
+            for (ConfirmListener l : this.confirmListeners) {
+                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 : this.confirmListeners) {
+                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/functional/Confirm.java

@@ -17,7 +17,6 @@
 
 package com.rabbitmq.client.test.functional;
 
-import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;

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/client/test/server/Firehose.java

@@ -54,6 +54,7 @@ private GetResponse publishGet(String key) throws IOException {
 
     private void checkHeaders(Map<String, Object> pubHeaders) {
         assertEquals("test", pubHeaders.get("exchange_name").toString());
+        @SuppressWarnings("unchecked")
         List<Object> routing_keys = (List<Object>) pubHeaders.get("routing_keys");
         assertEquals("traced", routing_keys.get(0).toString());
     }

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

@@ -117,11 +117,13 @@ public static void main(String[] args) {
             }
             Thread[] producerThreads = new Thread[producerCount];
             Connection[] producerConnections = new Connection[producerCount];
+            Channel[] producerChannels = new Channel[producerCount];
             for (int i = 0; i < producerCount; i++) {
                 System.out.println("starting producer #" + i);
                 Connection conn = factory.newConnection();
                 producerConnections[i] = conn;
                 Channel channel = conn.createChannel();
+                producerChannels[i] = channel;
                 if (producerTxSize > 0) channel.txSelect();
                 if (confirm >= 0) channel.confirmSelect();
                 channel.exchangeDeclare(exchangeName, exchangeType);
@@ -130,15 +132,17 @@ 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();
             }
 
             for (int i = 0; i < producerCount; i++) {
                 producerThreads[i].join();
+                producerChannels[i].clearReturnListeners();
+                producerChannels[i].clearConfirmListeners();
                 producerConnections[i].close();
             }