Merge with default

Author: Rob Harrop

codegen.py

@@ -696,4 +696,19 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * @return the sequence number of the next message to be published
      */
     long getNextPublishSeqNo();
+
+    /**
+     * Asynchronously send a method over this channel.
+     * @param method method to transmit over this channel.
+     * @throws IOException Problem transmitting method.
+     */
+    void asyncRpc(Method method) throws IOException;
+
+    /**
+     * Synchronously send a method over this channel.
+     * @param method method to transmit over this channel.
+     * @return response to method. Caller should cast as appropriate.
+     * @throws IOException Problem transmitting method.
+     */
+    Method rpc(Method method) throws IOException;
 }

src/com/rabbitmq/client/Channel.java

@@ -16,7 +16,6 @@
 
 package com.rabbitmq.client.impl;
 
-import java.io.IOException;
 import java.util.Date;
 import java.util.Map;
 import java.util.Hashtable;
@@ -32,8 +31,7 @@ public Object clone() throws CloneNotSupportedException {
 
         Map<String, Object> thisHeaders = getHeaders();
         if (thisHeaders != null) {
-            Map<String, Object> headers = new Hashtable<String, Object>();
-            headers.putAll(thisHeaders);
+            Map<String, Object> headers = new Hashtable<String, Object>(thisHeaders);
             bpClone.setHeaders(headers);
         }
 

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

@@ -218,8 +218,9 @@ public static void checkEmptyContentBodyFrameSize() {
         }
         int actualLength = s.toByteArray().length;
         if (EMPTY_CONTENT_BODY_FRAME_SIZE != actualLength) {
-            throw new AssertionError("Internal error: EMPTY_CONTENT_BODY_FRAME_SIZE is " + "incorrect - defined as " + EMPTY_CONTENT_BODY_FRAME_SIZE
-                    + ", where the computed value is in fact " + actualLength);
+            throw new AssertionError("Internal error: expected EMPTY_CONTENT_BODY_FRAME_SIZE("
+                    + EMPTY_CONTENT_BODY_FRAME_SIZE
+                    + ") is not equal to computed value: " + actualLength);
         }
     }
 

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

@@ -146,7 +146,7 @@ public void ensureIsOpen()
     private int _heartbeat;
 
     private final String _virtualHost;
-    private final int _requestedChannelMax, _requestedFrameMax, _requestedHeartbeat;
+//    private final int _requestedChannelMax, _requestedFrameMax, _requestedHeartbeat;
     private final Map<String, Object> _clientProperties;
 
     /** Saved server properties field from connection.start */
@@ -194,9 +194,6 @@ public AMQConnection(ConnectionFactory factory,
         checkPreconditions();
 
         _virtualHost = factory.getVirtualHost();
-        _requestedChannelMax = factory.getRequestedChannelMax();
-        _requestedFrameMax = factory.getRequestedFrameMax();
-        _requestedHeartbeat = factory.getRequestedHeartbeat();
         _clientProperties = new HashMap<String, Object>(factory.getClientProperties());
 
         _factory = factory;

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

@@ -124,35 +124,35 @@ private void addChannel(ChannelN chan) {
     }
 
     /**
-     * Remove the argument channel from the channel map. 
-     * This method must be safe to call multiple times on the same channel. If 
+     * 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. 
+        // 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 
+        // 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.   
+        // but it's much easier to just catch it here.
         synchronized (_channelMap) {
           ChannelN existing = _channelMap.remove(channelNumber);
-          // Nothing to do here. Move along. 
+          // 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. 
+          // and pretend we didn't touch it.
           else if (existing != channel) {
             _channelMap.put(channelNumber, existing);
             return;
           }
           channelNumberAllocator.free(channelNumber);
-        } 
+        }
     }
 }

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

@@ -242,7 +242,7 @@ public void releaseChannelNumber() {
             if (method instanceof Basic.Deliver) {
                 Basic.Deliver m = (Basic.Deliver) method;
 
-                Consumer callback = _consumers.get(m.consumerTag);
+                Consumer callback = _consumers.get(m.getConsumerTag());
                 if (callback == null) {
                     if (defaultConsumer == null) {
                         // No handler set. We should blow up as this message
@@ -257,20 +257,20 @@ public void releaseChannelNumber() {
                     }
                 }
 
-                Envelope envelope = new Envelope(m.deliveryTag,
-                                                 m.redelivered,
-                                                 m.exchange,
-                                                 m.routingKey);
+                Envelope envelope = new Envelope(m.getDeliveryTag(),
+                                                 m.getRedelivered(),
+                                                 m.getExchange(),
+                                                 m.getRoutingKey());
                 try {
-                    callback.handleDelivery(m.consumerTag,
+                    callback.handleDelivery(m.getConsumerTag(),
                                             envelope,
                                             (BasicProperties) command.getContentHeader(),
                                             command.getContentBody());
                 } catch (Throwable ex) {
                     _connection.getExceptionHandler().handleConsumerException(this,
                                                                               ex,
                                                                               callback,
-                                                                              m.consumerTag,
+                                                                              m.getConsumerTag(),
                                                                               "handleDelivery");
                 }
                 return true;
@@ -279,10 +279,10 @@ public void releaseChannelNumber() {
                 if (l != null) {
                     Basic.Return basicReturn = (Basic.Return) method;
                     try {
-                        l.handleReturn(basicReturn.replyCode,
-                                            basicReturn.replyText,
-                                            basicReturn.exchange,
-                                            basicReturn.routingKey,
+                        l.handleReturn(basicReturn.getReplyCode(),
+                                            basicReturn.getReplyText(),
+                                            basicReturn.getExchange(),
+                                            basicReturn.getRoutingKey(),
                                             (BasicProperties)
                                             command.getContentHeader(),
                                             command.getContentBody());
@@ -295,14 +295,14 @@ public void releaseChannelNumber() {
             } else if (method instanceof Channel.Flow) {
                 Channel.Flow channelFlow = (Channel.Flow) method;
                 synchronized (_channelMutex) {
-                    _blockContent = !channelFlow.active;
-                    transmit(new Channel.FlowOk(channelFlow.active));
+                    _blockContent = !channelFlow.getActive();
+                    transmit(new Channel.FlowOk(!_blockContent));
                     _channelMutex.notifyAll();
                 }
                 FlowListener l = getFlowListener();
                 if (l != null) {
                     try {
-                        l.handleFlow(channelFlow.active);
+                        l.handleFlow(channelFlow.getActive());
                     } catch (Throwable ex) {
                         _connection.getExceptionHandler().handleFlowListenerException(this, ex);
                     }
@@ -711,13 +711,13 @@ public GetResponse basicGet(String queue, boolean autoAck)
 
         if (method instanceof Basic.GetOk) {
             Basic.GetOk getOk = (Basic.GetOk)method;
-            Envelope envelope = new Envelope(getOk.deliveryTag,
-                                             getOk.redelivered,
-                                             getOk.exchange,
-                                             getOk.routingKey);
+            Envelope envelope = new Envelope(getOk.getDeliveryTag(),
+                                             getOk.getRedelivered(),
+                                             getOk.getExchange(),
+                                             getOk.getRoutingKey());
             BasicProperties props = (BasicProperties)replyCommand.getContentHeader();
             byte[] body = replyCommand.getContentBody();
-            int messageCount = getOk.messageCount;
+            int messageCount = getOk.getMessageCount();
             return new GetResponse(envelope, props, body, messageCount);
         } else if (method instanceof Basic.GetEmpty) {
             return null;
@@ -777,7 +777,7 @@ public String basicConsume(String queue, boolean autoAck, String consumerTag,
     {
         BlockingRpcContinuation<String> k = new BlockingRpcContinuation<String>() {
             public String transformReply(AMQCommand replyCommand) {
-                String actualConsumerTag = ((Basic.ConsumeOk) replyCommand.getMethod()).consumerTag;
+                String actualConsumerTag = ((Basic.ConsumeOk) replyCommand.getMethod()).getConsumerTag();
                 _consumers.put(actualConsumerTag, callback);
                 // We need to call back inside the connection thread
                 // in order avoid races with 'deliver' commands
@@ -896,7 +896,7 @@ public Confirm.SelectOk confirmSelect()
 
     /** Public API - {@inheritDoc} */
     public Channel.FlowOk flow(final boolean a) throws IOException {
-        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();
+        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow(a)).getMethod();
     }
 
     /** Public API - {@inheritDoc} */
@@ -908,5 +908,14 @@ public Channel.FlowOk getFlow() {
     public long getNextPublishSeqNo() {
         return nextPublishSeqNo;
     }
-    
+
+    public void asyncRpc(com.rabbitmq.client.Method method) throws IOException {
+        // This cast should eventually go
+        transmit((com.rabbitmq.client.impl.Method)method);
+    }
+
+    public com.rabbitmq.client.Method rpc(com.rabbitmq.client.Method method) throws IOException {
+        return exnWrappingRpc((com.rabbitmq.client.impl.Method)method).getMethod();
+    }
+
 }

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

@@ -19,8 +19,6 @@
 
 import java.io.DataOutputStream;
 import java.io.IOException;
-import java.math.BigDecimal;
-import java.math.BigInteger;
 import java.util.Date;
 import java.util.Map;
 

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

@@ -32,7 +32,7 @@ public class LongStringHelper
      * Private API - Implementation of {@link LongString}. When
      * interpreting bytes as a string, uses UTF-8 encoding.
      */
-    public static class ByteArrayLongString
+    private static class ByteArrayLongString
         implements LongString
     {
         byte [] bytes;

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

@@ -52,13 +52,6 @@ public abstract class Method implements com.rabbitmq.client.Method {
      */
     public abstract Object visit(MethodVisitor visitor) throws IOException;
 
-    /**
-     * Private API - Autogenerated reader for this method.
-     * @param reader interface to an object to read the method arguments
-     * @throws IOException if an error is encountered
-     */
-    public abstract void readArgumentsFrom(MethodArgumentReader reader) throws IOException;
-
     /**
      * Private API - Autogenerated writer for this method.
      * @param writer interface to an object to write the method arguments