Merge default into amqp_0_9_1

Author: Simon MacMullen

Makefile

@@ -7,7 +7,7 @@ SRC_ARCHIVE=$(PACKAGE_NAME)-$(VERSION)
 SIGNING_KEY=056E8E56
 GNUPG_PATH=~
 
-WEB_URL=http://stage.rabbitmq.com/
+WEB_URL=http://www.rabbitmq.com/
 NEXUS_STAGE_URL=http://oss.sonatype.org/service/local/staging/deploy/maven2
 
 AMQP_CODEGEN_DIR=$(shell fgrep sibling.codegen.dir build.properties | sed -e 's:sibling\.codegen\.dir=::')

src/com/rabbitmq/client/Channel.java

@@ -101,6 +101,11 @@ public interface Channel extends ShutdownNotifier {
      */
     FlowOk flow(boolean active) throws IOException;
 
+    /**
+     * Return the current Channel.Flow settings.
+     */
+    FlowOk getFlow();
+
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
      * and message 'OK'.
@@ -130,6 +135,52 @@ public interface Channel extends ShutdownNotifier {
      */
     void setReturnListener(ReturnListener listener);
 
+    /**
+     * Return the current {@link FlowListener}.
+     * @return an interface to the current flow listener.
+     */
+    FlowListener getFlowListener();
+
+    /**
+     * Set the current {@link FlowListener}.
+     * @param listener the listener to use, or null indicating "don't use one".
+     */
+    void setFlowListener(FlowListener listener);
+
+    /**
+     * Get the current default consumer. @see setDefaultConsumer for rationale.
+     * @return an interface to the current default consumer.
+     */
+    Consumer getDefaultConsumer();
+
+    /**
+     * Set the current default consumer.
+     *
+     * Under certain circumstances it is possible for a channel to receive a
+     * message delivery which does not match any consumer which is currently
+     * set up via basicConsume(). This will occur after the following sequence
+     * of events:
+     *
+     * ctag = basicConsume(queue, consumer); // i.e. with explicit acks
+     * // some deliveries take place but are not acked
+     * basicCancel(ctag);
+     * basicRecover(false);
+     *
+     * Since requeue is specified to be false in the basicRecover, the spec
+     * states that the message must be redelivered to "the original recipient"
+     * - i.e. the same channel / consumer-tag. But the consumer is no longer
+     * active.
+     *
+     * In these circumstances, you can register a default consumer to handle
+     * such deliveries. If no default consumer is registered an
+     * IllegalStateException will be thrown when such a delivery arrives.
+     *
+     * Most people will not need to use this.
+     *
+     * @param consumer the consumer to use, or null indicating "don't use one".
+     */
+    void setDefaultConsumer(Consumer consumer);
+
     /**
      * Request specific "quality of service" settings.
      *

src/com/rabbitmq/client/ConnectionFactory.java

@@ -335,7 +335,13 @@ protected FrameHandler createFrameHandler(Address addr)
         Socket socket = factory.createSocket();
         configureSocket(socket);
         socket.connect(new InetSocketAddress(hostName, portNumber));
-        return new SocketFrameHandler(socket);
+        return createFrameHandler(socket);
+    }
+
+    protected FrameHandler createFrameHandler(Socket sock)
+        throws IOException
+    {
+        return new SocketFrameHandler(sock);
     }
 
     /**

src/com/rabbitmq/client/Consumer.java

@@ -71,7 +71,15 @@ public interface Consumer {
     void handleShutdownSignal(String consumerTag, ShutdownSignalException sig);
 
     /**
+     * Called to notify the consumer that we've received a basic.recover-ok
+     * in reply to a basic.recover some other thread sent. All messages
+     * received before this is invoked that haven't been ack'ed will be
+     * redelivered. All messages received afterwards won't be.
      *
+     * This method exists since all the Consumer callbacks are invoked by the
+     * connection main loop thread - so it's sometimes useful to allow that
+     * thread to know that the recover-ok has been received, rather than the
+     * thread which invoked basicRecover().
      */
     void handleRecoverOk();
 

src/com/rabbitmq/client/FlowListener.java

@@ -0,0 +1,43 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+
+/**
+ * Implement this interface in order to be notified of Channel.Flow
+ * events.
+ */
+public interface FlowListener {
+    void handleFlow(boolean active)
+        throws IOException;
+}

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

@@ -37,6 +37,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.FlowListener;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.ReturnListener;
@@ -93,6 +94,15 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     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 default consumer, or null if there is
+     *  none.
+     */
+    public volatile Consumer defaultConsumer = null;
+
     /**
      * Construct a new channel on the given connection with the given
      * channel number. Usually not called directly - call
@@ -130,6 +140,32 @@ public void setReturnListener(ReturnListener listener) {
         returnListener = listener;
     }
 
+    /** Returns the current FlowListener. */
+    public FlowListener getFlowListener() {
+        return flowListener;
+    }
+
+    /**
+     * Sets the current FlowListener.
+     * A null argument is interpreted to mean "do not use a flow listener".
+     */
+    public void setFlowListener(FlowListener listener) {
+        flowListener = listener;
+    }
+
+    /** Returns the current default consumer. */
+    public Consumer getDefaultConsumer() {
+        return defaultConsumer;
+    }
+
+    /**
+     * Sets the current default consumer.
+     * A null argument is interpreted to mean "do not use a default consumer".
+     */
+    public void setDefaultConsumer(Consumer consumer) {
+        defaultConsumer = consumer;
+    }
+
     /**
      * Protected API - sends a ShutdownSignal to all active consumers.
      * @param signal an exception signalling channel shutdown
@@ -182,7 +218,7 @@ public void releaseChannelNumber() {
         // If we are not, however, then we are in a quiescing, or
         // shutting-down state as the result of an application
         // decision to close this channel, and we are to discard all
-        // incoming commands except for close and close-ok.
+        // incoming commands except for a close and close-ok.
 
         Method method = command.getMethod();
 
@@ -210,8 +246,17 @@ public void releaseChannelNumber() {
 
                 Consumer callback = _consumers.get(m.consumerTag);
                 if (callback == null) {
-                    // FIXME: what to do when we get such an unsolicited delivery?
-                    throw new UnsupportedOperationException("FIXME unsolicited delivery");
+                    if (defaultConsumer == null) {
+                        // No handler set. We should blow up as this message
+                        // needs acking, just dropping it is not enough. See bug
+                        // 22587 for discussion.
+                        throw new IllegalStateException("Unsolicited delivery -" +
+                                " see Channel.setDefaultConsumer to handle this" +
+                                " case.");
+                    }
+                    else {
+                        callback = defaultConsumer;
+                    }
                 }
 
                 Envelope envelope = new Envelope(m.deliveryTag,
@@ -256,6 +301,14 @@ public void releaseChannelNumber() {
                     transmit(new Channel.FlowOk(channelFlow.active));
                     _channelMutex.notifyAll();
                 }
+                FlowListener l = getFlowListener();
+                if (l != null) {
+                    try {
+                        l.handleFlow(channelFlow.active);
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleFlowListenerException(this, ex);
+                    }
+                }
                 return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
@@ -686,13 +739,14 @@ public Consumer transformReply(AMQCommand replyCommand) {
         }
     }
 
-    /** Public API - {@inheritDoc} */
+     /** Public API - {@inheritDoc} */
     public Basic.RecoverOk basicRecover(boolean requeue)
         throws IOException
     {
         return (Basic.RecoverOk) exnWrappingRpc(new Basic.Recover(requeue)).getMethod();
     }
 
+
     /** Public API - {@inheritDoc} */
     public void basicRecoverAsync(boolean requeue)
         throws IOException
@@ -725,4 +779,10 @@ public Tx.RollbackOk txRollback()
     public Channel.FlowOk flow(final boolean a) throws IOException {
         return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();
     }
+
+    /** Public API - {@inheritDoc} */
+    public Channel.FlowOk getFlow() {
+        return new Channel.FlowOk(!_blockContent);
+    }
+
 }

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

@@ -48,40 +48,40 @@ public void handleUnexpectedConnectionDriverException(Connection conn, Throwable
     }
 
     public void handleReturnListenerException(Channel channel, Throwable exception) {
-        // TODO: Convert to logging framework
-        System.err.println("ReturnListener.handleBasicReturn threw an exception for channel " +
-                           channel + ":");
-        exception.printStackTrace();
-        try {
-            ((AMQConnection) channel.getConnection()).close(AMQP.INTERNAL_ERROR,
-                                                            "Internal error in ReturnListener",
-                                                            false,
-                                                            exception);
-        } catch (IOException ioe) {
-            // Man, this clearly isn't our day.
-            // Ignore the exception? TODO: Log the nested failure
-        }
+        handleChannelKiller(channel, exception, "ReturnListener.handleBasicReturn");
     }
 
-    public void handleConsumerException(Channel channel,
-                                        Throwable exception,
-                                        Consumer consumer,
-                                        String consumerTag,
+    public void handleFlowListenerException(Channel channel, Throwable exception) {
+        handleChannelKiller(channel, exception, "FlowListener.handleFlow");
+    }
+
+    public void handleConsumerException(Channel channel, Throwable exception,
+                                        Consumer consumer, String consumerTag,
                                         String methodName)
+    {
+        handleChannelKiller(channel, exception, "Consumer " + consumer
+                                              + " (" + consumerTag + ")"
+                                              + " method " + methodName
+                                              + " for channel " + channel);
+    }
+
+    protected void handleChannelKiller(Channel channel,
+                                       Throwable exception,
+                                       String what)
     {
         // TODO: Convert to logging framework
-        System.err.println("Consumer " + consumer + " method " + methodName + " for channel " +
-                           channel + " threw an exception:");
+        System.err.println(what + " threw an exception for channel " +
+                           channel + ":");
         exception.printStackTrace();
         try {
             ((AMQConnection) channel.getConnection()).close(AMQP.INTERNAL_ERROR,
-                                                            "Internal error in Consumer " +
-                                                              consumerTag,
+                                                            "Internal error in " + what,
                                                             false,
                                                             exception);
         } catch (IOException ioe) {
             // Man, this clearly isn't our day.
             // Ignore the exception? TODO: Log the nested failure
         }
+
     }
 }

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

@@ -57,6 +57,16 @@ public interface ExceptionHandler {
      */
     void handleReturnListenerException(Channel channel, Throwable exception);
 
+    /**
+     * Perform any required exception processing for the situation
+     * when the driver thread for the connection has called a 
+     * FlowListener's handleFlow method, and that method has
+     * thrown an exeption.
+     * @param channel the ChannelN that held the FlowListener
+     * @param exception the exception thrown by FlowListener.handleFlow
+     */
+    void handleFlowListenerException(Channel channel, Throwable exception);
+
     /**
      * Perform any required exception processing for the situation
      * when the driver thread for the connection has called a method

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

@@ -197,6 +197,10 @@ public void handleReturnListenerException(Channel ch, Throwable ex) {
             fail("handleReturnListenerException: " + ex);
         }
 
+        public void handleFlowListenerException(Channel ch, Throwable ex) {
+            fail("handleFlowListenerException: " + ex);
+        }
+
         public void handleConsumerException(Channel ch,
                                             Throwable ex,
                                             Consumer c,

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

@@ -0,0 +1,19 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+/**
+ * See bug 21843. It's not obvious this is the right thing to do, but it's in
+ * the spec.
+ */
+public class BindToDefaultExchange extends BrokerTestCase {
+    public void testBindToDefaultExchange() throws IOException {
+        String queue = channel.queueDeclare().getQueue();
+        channel.queueBind(queue, "", "foobar");
+        basicPublishVolatile("", "foobar"); // Explicit binding
+        basicPublishVolatile("", queue); // Implicit binding
+        assertDelivered(queue, 2);
+    }
+}