Merge default into bug21949

Author: tonyg

src/com/rabbitmq/client/Channel.java

@@ -37,6 +37,7 @@
 import com.rabbitmq.client.AMQP.Exchange;
 import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.AMQP.Tx;
+import com.rabbitmq.client.impl.AMQImpl.Channel.FlowOk;
 
 /**
  * Public API: Interface to an AMQ channel. See the <a href="http://www.amqp.org/">spec</a> for details.
@@ -90,6 +91,14 @@ public interface Channel extends ShutdownNotifier {
      * @throws java.io.IOException if an error is encountered
      */
     void close(int closeCode, String closeMessage) throws IOException;
+    
+    /**
+     * Set flow on the channel
+     * 
+     * @param active if true, the server is asked to start sending. If false, the server is asked to stop sending.
+     * @throws IOException
+     */
+    FlowOk flow(boolean active) throws IOException;
 
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code

src/com/rabbitmq/client/Connection.java

@@ -103,6 +103,12 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      */
     Address[] getKnownHosts();
 
+    /**
+     * Retrieve the server properties.
+     * @return a map of the server properties. This typically includes the product name and version of the server.
+     */
+    Map<String, Object> getServerProperties();
+
     /**
      * Create a new channel, using an internally allocated channel number.
      * @return a new channel descriptor, or null if none is available

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

@@ -128,8 +128,8 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channelNumber the number of the channel to disconnect
      */
-    public final void disconnectChannel(int channelNumber) {
-        _channelManager.disconnectChannel(channelNumber);
+    public final void disconnectChannel(ChannelN channel) {
+        _channelManager.disconnectChannel(channel);
     }
 
     public void ensureIsOpen()
@@ -163,6 +163,9 @@ public void ensureIsOpen()
     private final int _requestedChannelMax, _requestedFrameMax, _requestedHeartbeat;
     private final Map<String, Object> _clientProperties;
 
+    /** Saved server properties field from connection.start */
+    public Map<String, Object> _serverProperties;
+
     /** {@inheritDoc} */
     public String getHost() {
         return _frameHandler.getHost();
@@ -182,6 +185,11 @@ public FrameHandler getFrameHandler(){
         return _frameHandler;
     }
 
+    /** {@inheritDoc} */
+    public Map<String, Object> getServerProperties() {
+        return _serverProperties;
+    }
+
     /**
      * Construct a new connection to a broker.
      * @param factory the initialization parameters for a connection
@@ -261,6 +269,8 @@ public void start(boolean insist)
         try {
             AMQP.Connection.Start connStart =
                 (AMQP.Connection.Start) connStartBlocker.getReply().getMethod();
+
+            _serverProperties = connStart.getServerProperties();
 
             Version serverVersion =
                 new Version(connStart.getVersionMajor(),

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

@@ -46,7 +46,8 @@
 
 public class ChannelManager {
     /** Mapping from channel number to AMQChannel instance */
-    private final Map<Integer, ChannelN> _channelMap = Collections.synchronizedMap(new HashMap<Integer, ChannelN>());
+    private final Map<Integer, ChannelN> _channelMap =
+        Collections.synchronizedMap(new HashMap<Integer, ChannelN>());
     private final IntAllocator channelNumberAllocator;
 
     /** Maximum channel number available on this connection. */
@@ -56,22 +57,22 @@ public int getChannelMax(){
       return _channelMax;
     }
 
-    public ChannelManager(int channelMax){
-      if(channelMax == 0){
-        // The framing encoding only allows for unsigned 16-bit integers for the channel number
-        channelMax = (1 << 16) - 1;
-      }
-
-      _channelMax = channelMax;
-      channelNumberAllocator = new IntAllocator(1, channelMax);
+    public ChannelManager(int channelMax) {
+        if (channelMax == 0) {
+            // The framing encoding only allows for unsigned 16-bit integers
+            // for the channel number
+            channelMax = (1 << 16) - 1;
+        }
+        _channelMax = channelMax;
+        channelNumberAllocator = new IntAllocator(1, channelMax);
     }
 
-    
+
     /**
      * Public API - Looks up an existing channel associated with this connection.
      * @param channelNumber the number of the required channel
      * @return the relevant channel descriptor
-     * @throws UnknownChannelException if there is no Channel associated with the 
+     * @throws UnknownChannelException if there is no Channel associated with the
      *         required channel number.
      */
     public ChannelN getChannel(int channelNumber) {
@@ -85,8 +86,8 @@ public void handleSignal(ShutdownSignalException signal) {
         synchronized(_channelMap) {
             channels = new HashSet<ChannelN>(_channelMap.values());
         }
-        for (AMQChannel channel : channels) {
-            disconnectChannel(channel.getChannelNumber());
+        for (ChannelN channel : channels) {
+            disconnectChannel(channel);
             channel.processShutdownSignal(signal, true, true);
         }
     }
@@ -100,7 +101,7 @@ public synchronized ChannelN createChannel(AMQConnection connection) throws IOEx
     }
 
     public synchronized ChannelN createChannel(AMQConnection connection, int channelNumber) throws IOException {
-        if(channelNumberAllocator.reserve(channelNumber)) 
+        if(channelNumberAllocator.reserve(channelNumber))
             return createChannelInternal(connection, channelNumber);
         else
             return null;
@@ -110,10 +111,10 @@ private synchronized ChannelN createChannelInternal(AMQConnection connection, in
         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."); 
+            // 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.");
         }
         ChannelN ch = new ChannelN(connection, channelNumber);
         addChannel(ch);
@@ -125,8 +126,36 @@ private void addChannel(ChannelN chan) {
         _channelMap.put(chan.getChannelNumber(), chan);
     }
 
-    public synchronized void disconnectChannel(int channelNumber) {
-        _channelMap.remove(channelNumber);
-        channelNumberAllocator.free(channelNumber);
+    /**
+     * Remove the argument channel from the channel map. 
+     * This method must be safe to call multiple times on the same channel. If 
+     * it is not then things go badly wrong.
+     */
+    public synchronized void disconnectChannel(ChannelN channel) {
+        int channelNumber = channel.getChannelNumber();
+       
+        // Warning, here be dragons. Not great big ones, but little baby ones
+        // which will nibble on your toes and occasionally trip you up when 
+        // you least expect it. 
+        // Basically, there's a race that can end us up here. It almost never 
+        // happens, but it's easier to repair it when it does than prevent it 
+        // from happening in the first place. 
+        // If we end up doing a Channel.close in one thread and a Channel.open
+        // with the same channel number in another, the two can overlap in such
+        // a way as to cause disconnectChannel on the old channel to try to 
+        // remove the new one. Ideally we would fix this race at the source,
+        // but it's much easier to just catch it here.   
+        synchronized (_channelMap) {
+          ChannelN existing = _channelMap.remove(channelNumber);
+          // Nothing to do here. Move along. 
+          if (existing == null) return;
+          // Oops, we've gone and stomped on someone else's channel. Put it back
+          // and pretend we didn't touch it. 
+          else if (existing != channel) {
+            _channelMap.put(channelNumber, existing);
+            return;
+          }
+          channelNumberAllocator.free(channelNumber);
+        } 
     }
 }

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

@@ -166,7 +166,7 @@ public void broadcastShutdownSignal(ShutdownSignalException signal) {
     }
 
     public void releaseChannelNumber() {
-        _connection.disconnectChannel(_channelNumber);
+        _connection.disconnectChannel(this);
     }
 
     /**
@@ -695,4 +695,9 @@ public Tx.RollbackOk txRollback()
     {
         return (Tx.RollbackOk) exnWrappingRpc(new Tx.Rollback()).getMethod();
     }
+
+    /** Public API - {@inheritDoc} */
+    public Channel.FlowOk flow(final boolean a) throws IOException {
+        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();        
+    }
 }

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

@@ -359,6 +359,19 @@ public void testLimitInheritsUnackedCount()
         drain(c, 1);
     }
 
+    public void testFlow() throws IOException
+    {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        declareBindConsume(c);
+        fill(1);
+        drain(c, 1);
+        channel.flow(false);
+        fill(1);
+        drain(c, 0);
+        channel.flow(true);
+        drain(c, 1);
+    }
+    
     protected void runLimitTests(int limit,
                                  boolean multiAck,
                                  boolean txMode,