Merge heads.

Author: Jerry Kuch

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/examples/HelloJsonClient.java

@@ -24,6 +24,8 @@
 import com.rabbitmq.tools.jsonrpc.JsonRpcClient;
 
 public class HelloJsonClient {
+    private static final int RPC_TIMEOUT_ONE_SECOND = 1000;
+
     public static void main(String[] args) {
         try {
             String request = (args.length > 0) ? args[0] : "Rabbit";
@@ -35,7 +37,7 @@ public static void main(String[] args) {
             cfconn.setPort(portNumber);
             Connection conn = cfconn.newConnection();
             Channel ch = conn.createChannel();
-            JsonRpcClient client = new JsonRpcClient(ch, "", "Hello");
+            JsonRpcClient client = new JsonRpcClient(ch, "", "Hello", RPC_TIMEOUT_ONE_SECOND);
             HelloJsonService service =
                 (HelloJsonService) client.createProxy(HelloJsonService.class);