Merge heads

Author: Rob Harrop

codegen.py

@@ -161,7 +161,6 @@ def printHeader():
         print
         print "import com.rabbitmq.client.impl.ContentHeaderPropertyWriter;"
         print "import com.rabbitmq.client.impl.ContentHeaderPropertyReader;"
-        print "import com.rabbitmq.client.impl.LongString;"
         print "import com.rabbitmq.client.impl.LongStringHelper;"
 
     def printProtocolClass():
@@ -422,6 +421,7 @@ def printHeader():
         print "import java.util.Map;"
         print
         print "import com.rabbitmq.client.AMQP;"
+        print "import com.rabbitmq.client.LongString;"
         print "import com.rabbitmq.client.UnknownClassOrMethodId;"
         print "import com.rabbitmq.client.UnexpectedMethodError;"
 

src/com/rabbitmq/client/Channel.java

@@ -697,6 +697,21 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     long getNextPublishSeqNo();
 
+    /**
+     * Wait until all messages published since the last call have been
+     * either ack'd or nack'd by the broker.  Note, when called on a
+     * non-Confirm channel, waitForConfirms returns true immediately.
+     * @return whether all the messages were ack'd (and none were nack'd)
+     */
+    boolean waitForConfirms() throws InterruptedException;
+
+    /** Wait until all messages published since the last call have
+     * been either ack'd or nack'd by the broker.  If any of the
+     * messages were nack'd, waitForConfirmsOrDie will throw an
+     * IOException.  When called on a non-Confirm channel, it will
+     * return immediately. */
+    void waitForConfirmsOrDie() throws IOException, InterruptedException;
+
     /**
      * Asynchronously send a method over this channel.
      * @param method method to transmit over this channel.
@@ -707,8 +722,8 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
     /**
      * Synchronously send a method over this channel.
      * @param method method to transmit over this channel.
-     * @return response to method. Caller should cast as appropriate.
+     * @return command response to method. Caller should cast as appropriate.
      * @throws IOException Problem transmitting method.
      */
-    Method rpc(Method method) throws IOException;
+    Command rpc(Method method) throws IOException;
 }

src/com/rabbitmq/client/Command.java

@@ -17,9 +17,6 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.Method;
-
-
 /**
  * Interface to a container for an AMQP method-and-arguments, with optional content header and body.
  */

src/com/rabbitmq/client/Connection.java

@@ -116,7 +116,7 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      * @throws IOException if an I/O problem is encountered
      */
     void close() throws IOException;
-    
+
     /**
      * Close this connection and all its channels.
      *

src/com/rabbitmq/client/JDKSaslConfig.java

@@ -16,7 +16,6 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.LongString;
 import com.rabbitmq.client.impl.LongStringHelper;
 
 import javax.security.auth.callback.Callback;

src/com/rabbitmq/client/impl/LongString.java renamed to src/com/rabbitmq/client/LongString.java

@@ -15,7 +15,7 @@
 //
 
 
-package com.rabbitmq.client.impl;
+package com.rabbitmq.client;
 
 import java.io.DataInputStream;
 import java.io.IOException;
@@ -31,26 +31,26 @@ public interface LongString
     public static final long MAX_LENGTH = 0xffffffffL;
 
     /**
-     * Get the length of the content of the long string in bytes
-     * @return the length in bytes >= 0 <= MAX_LENGTH
+     * @return the length of the {@link LongString} in bytes >= 0 <= MAX_LENGTH
      */
     public long length();
 
     /**
      * Get the content stream.
      * Repeated calls to this function return the same stream,
      * which may not support rewind.
-     * @return An input stream the reads the content
+     * @return An input stream that reads the content of the {@link LongString}
      * @throws IOException if an error is encountered
      */
     public DataInputStream getStream() throws IOException;
 
     /**
-     * Get the content as a byte array.
-     * Repeated calls to this function return the same array.
-     * This function will fail if getContentLength() > Integer.MAX_VALUE
-     * throwing an IllegalStateException.
-     * @return the content as an array
+     * Get the content as a byte array.  This need not be a copy. Updates to the
+     * returned array may change the value of the {@link LongString}.
+     * Repeated calls to this function may return the same array.
+     * This function will fail if <code><b>this</b>.length() > Integer.MAX_VALUE</code>,
+     * throwing an {@link IllegalStateException}.
+     * @return the array of bytes containing the content of the {@link LongString}
      */
     public byte [] getBytes();
 }

src/com/rabbitmq/client/RpcClient.java

@@ -234,7 +234,7 @@ public String stringCall(String message)
      * Perform an AMQP wire-protocol-table based RPC roundtrip <br><br>
      *
      * There are some restrictions on the values appearing in the table: <br>
-     * they must be of type {@link String}, {@link com.rabbitmq.client.impl.LongString}, {@link Integer}, {@link java.math.BigDecimal}, {@link Date},
+     * they must be of type {@link String}, {@link LongString}, {@link Integer}, {@link java.math.BigDecimal}, {@link Date},
      * or (recursively) a {@link Map} of the enclosing type.
      *
      * @param message the table to send

src/com/rabbitmq/client/SaslMechanism.java

@@ -16,7 +16,6 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.LongString;
 
 import java.io.IOException;
 

src/com/rabbitmq/client/UnexpectedMethodError.java

@@ -17,14 +17,13 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.Method;
-
 /**
  * Indicates that a {@link Method} object was supplied that was not
  * expected. For instance, {@link Channel#basicGet} throws this if it
  * receives anything other than {@link AMQP.Basic.GetOk} or
- * {@link AMQP.Basic.GetEmpty}, and {@link com.rabbitmq.client.impl.AMQImpl.DefaultMethodVisitor}
- * throws this as the default action within each visitor case.
+ * {@link AMQP.Basic.GetEmpty}, and the
+ * {@link com.rabbitmq.client.impl.AMQImpl.DefaultMethodVisitor}
+ * throws this as the action within each visitor case.
  */
 public class UnexpectedMethodError extends Error {
     private static final long serialVersionUID = 1L;

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

@@ -22,14 +22,17 @@
 
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Command;
+import com.rabbitmq.client.Method;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.utility.BlockingValueOrException;
 
 /**
- * Base class modelling an AMQ channel. Subclasses implement close()
- * and processAsync(), and may choose to override
- * processShutdownSignal().
+ * Base class modelling an AMQ channel. Subclasses implement
+ * {@link com.rabbitmq.client.Channel#close} and
+ * {@link #processAsync processAsync()}, and may choose to override
+ * {@link #processShutdownSignal processShutdownSignal()} and
+ * {@link #rpc rpc()}.
  *
  * @see ChannelN
  * @see Connection
@@ -118,11 +121,11 @@ public static IOException wrap(ShutdownSignalException ex, String message) {
     /**
      * Placeholder until we address bug 15786 (implementing a proper exception hierarchy).
      */
-    public AMQCommand exnWrappingRpc(com.rabbitmq.client.Method m)
+    public AMQCommand exnWrappingRpc(Method m)
         throws IOException
     {
         try {
-            return rpc((com.rabbitmq.client.impl.Method)m);
+            return privateRpc(m);
         } catch (AlreadyClosedException ace) {
             // Do not wrap it since it means that connection/channel
             // was closed in some action in the past
@@ -183,12 +186,18 @@ public void ensureIsOpen()
     }
 
     /**
-     * Protected API - sends a Command to the broker and waits for the
-     * next inbound Command from the broker: only for use from
+     * Protected API - sends a {@link Method} to the broker and waits for the
+     * next in-bound Command from the broker: only for use from
      * non-connection-MainLoop threads!
      */
     public AMQCommand rpc(Method m)
         throws IOException, ShutdownSignalException
+    {
+        return privateRpc(m);
+    }
+
+    private AMQCommand privateRpc(Method m)
+        throws IOException, ShutdownSignalException
     {
         SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
         rpc(m, k);
@@ -246,7 +255,7 @@ public void processShutdownSignal(ShutdownSignalException signal,
         try {
             synchronized (_channelMutex) {
                 if (!setShutdownCauseIfOpen(signal)) {
-                    if (!ignoreClosed) 
+                    if (!ignoreClosed)
                         throw new AlreadyClosedException("Attempt to use closed channel", this);
                 }