Merge bug24229 to default

Author: Simon MacMullen

codegen.py

@@ -370,6 +370,9 @@ def printSetter(fieldType, fieldName):
 
         print "        }"
 
+        # default constructor
+        print "        public %sProperties() {}" % (jClassName)
+
         #class properties
         print "        public int getClassId() { return %i; }" % (c.index)
         print "        public String getClassName() { return \"%s\"; }" % (c.name)

src/com/rabbitmq/client/RpcClient.java

@@ -27,6 +27,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.impl.MethodArgumentReader;
 import com.rabbitmq.client.impl.MethodArgumentWriter;
@@ -45,6 +46,10 @@ public class RpcClient {
     private final String _exchange;
     /** Routing key to use for requests */
     private final String _routingKey;
+    /** timeout to use on call responses */
+    private final int _timeout;
+    /** NO_TIMEOUT value must match convention on {@link BlockingCell#uninterruptibleGet(int)} */
+    protected final static int NO_TIMEOUT = -1;
 
     /** Map from request correlation ID to continuation BlockingCell */
     private final Map<String, BlockingCell<Object>> _continuationMap = new HashMap<String, BlockingCell<Object>>();
@@ -59,24 +64,44 @@ public class RpcClient {
     /**
      * Construct a new RpcClient that will communicate on the given channel, sending
      * requests to the given exchange with the given routing key.
-     * <p>
+     * <p/>
      * Causes the creation of a temporary private autodelete queue.
      * @param channel the channel to use for communication
      * @param exchange the exchange to connect to
      * @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) throws IOException {
+    public RpcClient(Channel channel, String exchange, String routingKey, int timeout) throws IOException {
         _channel = channel;
         _exchange = exchange;
         _routingKey = routingKey;
+        if (timeout < NO_TIMEOUT) throw new IllegalArgumentException("Timeout arguument must be NO_TIMEOUT(-1) or non-negative.");
+        _timeout = timeout;
         _correlationId = 0;
 
         _replyQueue = setupReplyQueue();
         _consumer = setupConsumer();
     }
 
+    /**
+     * Construct a new RpcClient that will communicate on the given channel, sending
+     * requests to the given exchange with the given routing key.
+     * <p/>
+     * Causes the creation of a temporary private autodelete queue.
+     * <p/>
+     * Waits forever for responses (that is, no timeout).
+     * @param channel the channel to use for communication
+     * @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);
+    }
+
     /**
      * Private API - ensures the RpcClient is correctly open.
      * @throws IOException if an error is encountered
@@ -151,7 +176,7 @@ public void publish(AMQP.BasicProperties props, byte[] message)
     }
 
     public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
-        throws IOException, ShutdownSignalException
+        throws IOException, ShutdownSignalException, TimeoutException
     {
         checkConsumer();
         BlockingCell<Object> k = new BlockingCell<Object>();
@@ -163,7 +188,7 @@ public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
             _continuationMap.put(replyId, k);
         }
         publish(props, message);
-        Object reply = k.uninterruptibleGet();
+        Object reply = k.uninterruptibleGet(_timeout);
         if (reply instanceof ShutdownSignalException) {
             ShutdownSignalException sig = (ShutdownSignalException) reply;
             ShutdownSignalException wrapper =
@@ -184,9 +209,10 @@ public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
      * @return the byte array response received
      * @throws ShutdownSignalException if the connection dies during our wait
      * @throws IOException if an error is encountered
+     * @throws TimeoutException if a response is not received within the configured timeout
      */
     public byte[] primitiveCall(byte[] message)
-        throws IOException, ShutdownSignalException {
+        throws IOException, ShutdownSignalException, TimeoutException {
         return primitiveCall(null, message);
     }
 
@@ -196,9 +222,10 @@ public byte[] primitiveCall(byte[] message)
      * @return the string response received
      * @throws ShutdownSignalException if the connection dies during our wait
      * @throws IOException if an error is encountered
+     * @throws TimeoutException if a timeout occurs before the response is received
      */
     public String stringCall(String message)
-        throws IOException, ShutdownSignalException
+        throws IOException, ShutdownSignalException, TimeoutException
     {
         return new String(primitiveCall(message.getBytes()));
     }
@@ -214,9 +241,10 @@ public String stringCall(String message)
      * @return the table received
      * @throws ShutdownSignalException if the connection dies during our wait
      * @throws IOException if an error is encountered
+     * @throws TimeoutException if a timeout occurs before a response is received
      */
     public Map<String, Object> mapCall(Map<String, Object> message)
-        throws IOException, ShutdownSignalException
+        throws IOException, ShutdownSignalException, TimeoutException
     {
         ByteArrayOutputStream buffer = new ByteArrayOutputStream();
         MethodArgumentWriter writer = new MethodArgumentWriter(new DataOutputStream(buffer));
@@ -239,9 +267,10 @@ public Map<String, Object> mapCall(Map<String, Object> message)
      * @return the table received
      * @throws ShutdownSignalException if the connection dies during our wait
      * @throws IOException if an error is encountered
+     * @throws TimeoutException if a timeout occurs before a response is received
      */
     public Map<String, Object> mapCall(Object[] keyValuePairs)
-        throws IOException, ShutdownSignalException
+        throws IOException, ShutdownSignalException, TimeoutException
     {
         Map<String, Object> message = new HashMap<String, Object>();
         for (int i = 0; i < keyValuePairs.length; i += 2) {

src/com/rabbitmq/tools/jsonrpc/JsonRpcClient.java

@@ -22,6 +22,7 @@
 import java.lang.reflect.Proxy;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.RpcClient;
@@ -57,20 +58,27 @@
  */
 public class JsonRpcClient extends RpcClient implements InvocationHandler {
     /** Holds the JSON-RPC service description for this client. */
-    public ServiceDescription serviceDescription;
+    private ServiceDescription serviceDescription;
 
     /**
      * Construct a new JsonRpcClient, passing the parameters through
      * to RpcClient's constructor. The service description record is
      * retrieved from the server during construction.
+     * @throws TimeoutException if a response is not received within the timeout specified, if any
      */
-    public JsonRpcClient(Channel channel, String exchange, String routingKey)
-        throws IOException, JsonRpcException
+    public JsonRpcClient(Channel channel, String exchange, String routingKey, int timeout)
+        throws IOException, JsonRpcException, TimeoutException
     {
-	super(channel, exchange, routingKey);
+	super(channel, exchange, routingKey, timeout);
 	retrieveServiceDescription();
     }
 
+    public JsonRpcClient(Channel channel, String exchange, String routingKey)
+    throws IOException, JsonRpcException, TimeoutException
+    {
+        this(channel, exchange, routingKey, RpcClient.NO_TIMEOUT);
+    }
+
     /**
      * Private API - parses a JSON-RPC reply object, checking it for exceptions.
      * @return the result contained within the reply, if no exception is found
@@ -96,27 +104,25 @@ public static Object checkReply(Map<String, Object> reply)
      * waits for the response.
      * @return the result contained within the reply, if no exception is found
      * @throws JsonRpcException if the reply object contained an exception
+     * @throws TimeoutException if a response is not received within the timeout specified, if any
      */
-    public Object call(String method, Object[] params)
-	throws IOException, JsonRpcException
+    public Object call(String method, Object[] params) throws IOException, JsonRpcException, TimeoutException
     {
-	HashMap<String, Object> request = new HashMap<String, Object>();
-	request.put("id", null);
-	request.put("method", method);
-	request.put("version", ServiceDescription.JSON_RPC_VERSION);
-	request.put("params", (params == null) ? new Object[0] : params);
+        HashMap<String, Object> request = new HashMap<String, Object>();
+        request.put("id", null);
+        request.put("method", method);
+        request.put("version", ServiceDescription.JSON_RPC_VERSION);
+        request.put("params", (params == null) ? new Object[0] : params);
         String requestStr = new JSONWriter().write(request);
-	String replyStr;
         try {
-            replyStr = this.stringCall(requestStr);
+            String replyStr = this.stringCall(requestStr);
+            @SuppressWarnings("unchecked")
+            Map<String, Object> map = (Map<String, Object>) (new JSONReader().read(replyStr));
+            return checkReply(map);
         } catch(ShutdownSignalException ex) {
             throw new IOException(ex.getMessage()); // wrap, re-throw
         }
 
-        //System.out.println(requestStr + " --->\n---> " + replyStr);
-	@SuppressWarnings("unchecked")
-    Map<String, Object> map = (Map<String, Object>) (new JSONReader().read(replyStr));
-	return checkReply(map);
     }
 
     /**
@@ -176,10 +182,11 @@ public static Object coerce(String val, String type)
      * @return the result contained within the reply, if no exception is found
      * @throws JsonRpcException if the reply object contained an exception
      * @throws NumberFormatException if a coercion failed
+     * @throws TimeoutException if a response is not received within the timeout specified, if any
      * @see #coerce
      */
     public Object call(String[] args)
-	throws NumberFormatException, IOException, JsonRpcException
+	throws NumberFormatException, IOException, JsonRpcException, TimeoutException
     {
 	if (args.length == 0) {
 	    throw new IllegalArgumentException("First string argument must be method name");
@@ -210,13 +217,13 @@ public ServiceDescription getServiceDescription() {
      * Private API - invokes the "system.describe" method on the
      * server, and parses and stores the resulting service description
      * in this object.
+     * TODO: Avoid calling this from the constructor.
+     * @throws TimeoutException if a response is not received within the timeout specified, if any
      */
-    public void retrieveServiceDescription()
-	throws IOException, JsonRpcException
+    private void retrieveServiceDescription() throws IOException, JsonRpcException, TimeoutException
     {
-	@SuppressWarnings("unchecked")
-    Map<String, Object> rawServiceDescription = (Map<String, Object>) call("system.describe", null);
-	//System.out.println(new JSONWriter().write(rawServiceDescription));
-	this.serviceDescription = new ServiceDescription(rawServiceDescription);
+        @SuppressWarnings("unchecked")
+        Map<String, Object> rawServiceDescription = (Map<String, Object>) call("system.describe", null);
+        serviceDescription = new ServiceDescription(rawServiceDescription);
     }
 }

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

@@ -111,16 +111,6 @@ protected void cleanup() throws IOException {
         }
     }
 
-    protected void publish(String key, boolean mandatory, boolean immediate)
-        throws IOException {
-        channel.basicPublish("x", key, mandatory, immediate, null,
-                             "ae-test".getBytes());
-    }
-
-    protected void publish(String key) throws IOException {
-        publish(key, false, false);
-    }
-
     /**
      * Perform an auto-acking 'basic.get' on each of the queues named
      * in {@link #resources} and check whether a message can be
@@ -161,7 +151,8 @@ protected void check(String key, boolean mandatory, boolean immediate,
         throws IOException {
 
         gotReturn.set(false);
-        publish(key, mandatory, immediate);
+        channel.basicPublish("x", key, mandatory, immediate, null,
+                             "ae-test".getBytes());
         checkGet(expected);
         assertEquals(ret, gotReturn.get());
     }
@@ -212,19 +203,6 @@ public void testAe() throws IOException {
             check(k, false, true, unrouted, true);
         }
 
-        //tx
-        channel.txSelect();
-        for (String k : keys) {
-            publish(k);
-            checkGet(unrouted);
-            channel.txRollback();
-            checkGet(unrouted);
-            publish(k);
-            checkGet(unrouted);
-            channel.txCommit();
-            checkGet(expected(k));
-        }
-
         cleanup();
     }