Prototype user of fast RPC replies.

Author: Simon MacMullen

src/com/rabbitmq/client/RpcClient.java

@@ -58,8 +58,6 @@ public class RpcClient {
     /** Contains the most recently-used request correlation ID */
     private int _correlationId;
 
-    /** The name of our private reply queue */
-    private String _replyQueue;
     /** Consumer attached to our reply queue */
     private DefaultConsumer _consumer;
 
@@ -73,7 +71,6 @@ public class RpcClient {
      * @param routingKey the routing key
      * @param timeout milliseconds before timing out on wait for response
      * @throws IOException if an error is encountered
-     * @see #setupReplyQueue
      */
     public RpcClient(Channel channel, String exchange, String routingKey, int timeout) throws IOException {
         _channel = channel;
@@ -83,7 +80,6 @@ public RpcClient(Channel channel, String exchange, String routingKey, int timeou
         _timeout = timeout;
         _correlationId = 0;
 
-        _replyQueue = setupReplyQueue();
         _consumer = setupConsumer();
     }
 
@@ -98,7 +94,6 @@ public RpcClient(Channel channel, String exchange, String routingKey, int timeou
      * @param exchange the exchange to connect to
      * @param routingKey the routing key
      * @throws IOException if an error is encountered
-     * @see #setupReplyQueue
      */
     public RpcClient(Channel channel, String exchange, String routingKey) throws IOException {
         this(channel, exchange, routingKey, NO_TIMEOUT);
@@ -125,16 +120,6 @@ public void close() throws IOException {
         }
     }
 
-    /**
-     * Creates a server-named exclusive autodelete queue to use for
-     * receiving replies to RPC requests.
-     * @throws IOException if an error is encountered
-     * @return the name of the reply queue
-     */
-    protected String setupReplyQueue() throws IOException {
-        return _channel.queueDeclare("", false, true, true, null).getQueue();
-    }
-
     /**
      * Registers a consumer on the reply queue.
      * @throws IOException if an error is encountered
@@ -167,7 +152,7 @@ public void handleDelivery(String consumerTag,
                 }
             }
         };
-        _channel.basicConsume(_replyQueue, true, consumer);
+        _channel.basicConsume("amq.reply-to", true, consumer);
         return consumer;
     }
 
@@ -186,7 +171,7 @@ public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
             _correlationId++;
             String replyId = "" + _correlationId;
             props = ((props==null) ? new AMQP.BasicProperties.Builder() : props.builder())
-                    .correlationId(replyId).replyTo(_replyQueue).build();
+                    .correlationId(replyId).replyTo("amq.reply-consumer").build();
             _continuationMap.put(replyId, k);
         }
         publish(props, message);
@@ -332,14 +317,6 @@ public int getCorrelationId() {
         return _correlationId;
     }
 
-    /**
-     * Retrieve the reply queue.
-     * @return the name of the client's reply queue
-     */
-    public String getReplyQueue() {
-        return _replyQueue;
-    }
-
     /**
      * Retrieve the consumer.
      * @return an interface to the client's consumer object