Merged bug 22101 into default

Author: Simon MacMullen

Makefile

@@ -1,7 +1,5 @@
 VERSION=0.0.0
-
 PACKAGE_NAME=rabbitmq-java-client
-
 JAVADOC_ARCHIVE=$(PACKAGE_NAME)-javadoc-$(VERSION)
 SRC_ARCHIVE=$(PACKAGE_NAME)-$(VERSION)
 SIGNING_KEY=056E8E56

src/com/rabbitmq/client/Channel.java

@@ -301,6 +301,56 @@ Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable
      */
     Exchange.DeleteOk exchangeDelete(String exchange) throws IOException;
 
+    /**
+     * Bind an exchange to an exchange, with no extra arguments.
+     * @see com.rabbitmq.client.AMQP.Exchange.Bind
+     * @see com.rabbitmq.client.AMQP.Exchange.BindOk
+     * @param destination: the name of the exchange to which messages flow across the binding
+     * @param source: the name of the exchange from which messages flow across the binding
+     * @param routingKey: the routine key to use for the binding
+     * @return a binding-confirm method if the binding was successfully created
+     * @throws java.io.IOException if an error is encountered
+     */
+    Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException;
+
+    /**
+     * Bind an exchange to an exchange.
+     * @see com.rabbitmq.client.AMQP.Exchange.Bind
+     * @see com.rabbitmq.client.AMQP.Exchange.BindOk
+     * @param destination: the name of the exchange to which messages flow across the binding
+     * @param source: the name of the exchange from which messages flow across the binding
+     * @param routingKey: the routine key to use for the binding
+     * @param arguments: other properties (binding parameters)
+     * @return a binding-confirm method if the binding was successfully created
+     * @throws java.io.IOException if an error is encountered
+     */
+    Exchange.BindOk exchangeBind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
+
+    /**
+     * Unbind an exchange from an exchange, with no extra arguments.
+     * @see com.rabbitmq.client.AMQP.Exchange.Bind
+     * @see com.rabbitmq.client.AMQP.Exchange.BindOk
+     * @param destination: the name of the exchange to which messages flow across the binding
+     * @param source: the name of the exchange from which messages flow across the binding
+     * @param routingKey: the routine key to use for the binding
+     * @return a binding-confirm method if the binding was successfully created
+     * @throws java.io.IOException if an error is encountered
+     */
+    Exchange.UnbindOk exchangeUnbind(String destination, String source, String routingKey) throws IOException;
+
+    /**
+     * Unbind an exchange from an exchange.
+     * @see com.rabbitmq.client.AMQP.Exchange.Bind
+     * @see com.rabbitmq.client.AMQP.Exchange.BindOk
+     * @param destination: the name of the exchange to which messages flow across the binding
+     * @param source: the name of the exchange from which messages flow across the binding
+     * @param routingKey: the routine key to use for the binding
+     * @param arguments: other properties (binding parameters)
+     * @return a binding-confirm method if the binding was successfully created
+     * @throws java.io.IOException if an error is encountered
+     */
+    Exchange.UnbindOk exchangeUnbind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
+
     /**
      * Actively declare a server-named exclusive, autodelete, non-durable queue.
      * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.

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

@@ -37,6 +37,9 @@
 import java.util.Map;
 import java.util.HashMap;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
@@ -89,7 +92,7 @@ public static Map<String, Object> defaultClientProperties() {
         new Version(AMQP.PROTOCOL.MAJOR, AMQP.PROTOCOL.MINOR);
 
     /** Initialization parameters */
-    private final ConnectionFactory factory;
+    private final ConnectionFactory _factory;
 
     /** The special channel 0 */
     private final AMQChannel _channel0 = new AMQChannel(this, 0) {
@@ -122,6 +125,9 @@ public static Map<String, Object> defaultClientProperties() {
     /** Flag indicating whether the client received Connection.Close message from the broker */
     private boolean _brokerInitiatedShutdown = false;
 
+    /** Manages heartbeat sending for this connection */
+    private final HeartbeatSender _heartbeatSender;
+
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channel the channel to disconnect
@@ -138,12 +144,6 @@ public void ensureIsOpen()
         }
     }
 
-    /**
-     * Timestamp of last time we wrote a frame - used for deciding when to
-     * send a heartbeat
-     */
-    private volatile long _lastActivityTime = Long.MAX_VALUE;
-
     /**
      * Count of socket-timeouts that have happened without any incoming frames
      */
@@ -210,7 +210,8 @@ public AMQConnection(ConnectionFactory factory,
         _requestedHeartbeat = factory.getRequestedHeartbeat();
         _clientProperties = new HashMap<String, Object>(factory.getClientProperties());
 
-        this.factory = factory;
+        _factory = factory;
+        _heartbeatSender = new HeartbeatSender(frameHandler);
         _frameHandler = frameHandler;
         _running = true;
         _frameMax = 0;
@@ -288,17 +289,17 @@ public void start()
         }
 
         int channelMax =
-            negotiatedMaxValue(factory.getRequestedChannelMax(),
+            negotiatedMaxValue(_factory.getRequestedChannelMax(),
                                connTune.getChannelMax());
         _channelManager = new ChannelManager(channelMax);
 
         int frameMax =
-            negotiatedMaxValue(factory.getRequestedFrameMax(),
+            negotiatedMaxValue(_factory.getRequestedFrameMax(),
                                connTune.getFrameMax());
         setFrameMax(frameMax);
 
         int heartbeat =
-            negotiatedMaxValue(factory.getRequestedHeartbeat(),
+            negotiatedMaxValue(_factory.getRequestedHeartbeat(),
                                connTune.getHeartbeat());
         setHeartbeat(heartbeat);
 
@@ -349,10 +350,12 @@ public int getHeartbeat() {
      */
     public void setHeartbeat(int heartbeat) {
         try {
+            _heartbeatSender.setHeartbeat(heartbeat);
+            _heartbeat = heartbeat;
+
             // Divide by four to make the maximum unwanted delay in
             // sending a timeout be less than a quarter of the
             // timeout setting.
-            _heartbeat = heartbeat;
             _frameHandler.setTimeout(heartbeat * 1000 / 4);
         } catch (SocketException se) {
             // should do more here?
@@ -395,7 +398,7 @@ public Frame readFrame() throws IOException {
      */
     public void writeFrame(Frame f) throws IOException {
         _frameHandler.writeFrame(f);
-        _lastActivityTime = System.nanoTime();
+        _heartbeatSender.signalActivity();
     }
 
     private static int negotiatedMaxValue(int clientValue, int serverValue) {
@@ -416,7 +419,7 @@ private class MainLoop extends Thread {
             try {
                 while (_running) {
                     Frame frame = readFrame();
-                    maybeSendHeartbeat();
+
                     if (frame != null) {
                         _missedHeartbeats = 0;
                         if (frame.type == AMQP.FRAME_HEARTBEAT) {
@@ -459,25 +462,6 @@ private class MainLoop extends Thread {
         }
     }
 
-    private static final long NANOS_IN_SECOND = 1000 * 1000 * 1000;
-
-    /**
-     * Private API - Checks lastActivityTime and heartbeat, sending a
-     * heartbeat frame if conditions are right.
-     */
-    public void maybeSendHeartbeat() throws IOException {
-        if (_heartbeat == 0) {
-            // No heartbeating.
-            return;
-        }
-
-        long now = System.nanoTime();
-        if (now > (_lastActivityTime + (_heartbeat * NANOS_IN_SECOND))) {
-            _lastActivityTime = now;
-            writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
-        }
-    }
-
     /**
      * Private API - Called when a frame-read operation times out. Checks to
      * see if too many heartbeats have been missed, and if so, throws
@@ -557,7 +541,7 @@ public void handleConnectionClose(Command closeCommand) {
         } catch (IOException ioe) {
             Utility.emptyStatement();
         }
-        _heartbeat = 0; // Do not try to send heartbeats after CloseOk
+        _heartbeatSender.shutdown(); // Do not try to send heartbeats after CloseOk
         _brokerInitiatedShutdown = true;
         Thread scw = new SocketCloseWait(sse);
         scw.setName("AMQP Connection Closing Monitor " +
@@ -607,6 +591,10 @@ public ShutdownSignalException shutdown(Object reason,
             if (isOpen())
                 _shutdownCause = sse;
         }
+
+        // stop any heartbeating
+        _heartbeatSender.shutdown();
+
         _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
         _channelManager.handleSignal(sse);
         return sse;

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

@@ -92,32 +92,43 @@ public void handleSignal(ShutdownSignalException signal) {
         }
     }
 
-    public synchronized ChannelN createChannel(AMQConnection connection) throws IOException {
-        int channelNumber = channelNumberAllocator.allocate();
-        if (channelNumber == -1) {
-            return null;
+    public ChannelN createChannel(AMQConnection connection) throws IOException {
+        int channelNumber;
+        synchronized (this) {
+            channelNumber = channelNumberAllocator.allocate();
+            if (channelNumber == -1) {
+                return null;
+            }
         }
         return createChannelInternal(connection, channelNumber);
     }
 
-    public synchronized ChannelN createChannel(AMQConnection connection, int channelNumber) throws IOException {
-        if(channelNumberAllocator.reserve(channelNumber))
+    public ChannelN createChannel(AMQConnection connection, int channelNumber) throws IOException {
+        boolean reserved;
+        synchronized (this) {
+            reserved = channelNumberAllocator.reserve(channelNumber);
+        }
+        if(reserved)
             return createChannelInternal(connection, channelNumber);
         else
             return null;
     }
 
-    private synchronized ChannelN createChannelInternal(AMQConnection connection, int channelNumber) throws IOException {
-        if (_channelMap.containsKey(channelNumber)) {
-            // That number's already allocated! Can't do it
-            // This should never happen unless something has gone
-            // badly wrong with our implementation.
-            throw new IllegalStateException("We have attempted to "
-              + "create a channel with a number that is already in "
-              + "use. This should never happen. Please report this as a bug.");
+    private ChannelN createChannelInternal(AMQConnection connection, int channelNumber) throws IOException {
+        ChannelN ch;
+        synchronized (this) {
+            if (_channelMap.containsKey(channelNumber)) {
+                // That number's already allocated! Can't do it
+                // This should never happen unless something has gone
+                // badly wrong with our implementation.
+                throw new IllegalStateException("We have attempted to "
+                        + "create a channel with a number that is already in "
+                        + "use. This should never happen. "
+                        + "Please report this as a bug.");
+            }
+            ch = new ChannelN(connection, channelNumber);
+            addChannel(ch);
         }
-        ChannelN ch = new ChannelN(connection, channelNumber);
-        addChannel(ch);
         ch.open(); // now that it's been added to our internal tables
         return ch;
     }

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

@@ -524,6 +524,36 @@ public Exchange.DeleteOk exchangeDelete(String exchange)
         return exchangeDelete(exchange, false);
     }
 
+    /** Public API - {@inheritDoc} */
+    public Exchange.BindOk exchangeBind(String destination, String source,
+            String routingKey, Map<String, Object> arguments)
+            throws IOException {
+        return (Exchange.BindOk) exnWrappingRpc(
+                new Exchange.Bind(TICKET, destination, source, routingKey,
+                        false, arguments)).getMethod();
+    }
+
+    /** Public API - {@inheritDoc} */
+    public Exchange.BindOk exchangeBind(String destination, String source,
+            String routingKey) throws IOException {
+        return exchangeBind(destination, source, routingKey, null);
+    }
+
+    /** Public API - {@inheritDoc} */
+    public Exchange.UnbindOk exchangeUnbind(String destination, String source,
+            String routingKey, Map<String, Object> arguments)
+            throws IOException {
+        return (Exchange.UnbindOk) exnWrappingRpc(
+                new Exchange.Unbind(TICKET, destination, source, routingKey,
+                        false, arguments)).getMethod();
+    }
+
+    /** Public API - {@inheritDoc} */
+    public Exchange.UnbindOk exchangeUnbind(String destination, String source,
+            String routingKey) throws IOException {
+        return exchangeUnbind(destination, source, routingKey, null);
+    }
+    
     /** Public API - {@inheritDoc} */
     public Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive,
                                         boolean autoDelete, Map<String, Object> arguments)