Merge pull request rabbitmq#115 from monosoul/updated-tutorial-six-java

Updated tutorial six Java code examples due to QueueingConsumer deprecation

Author: acogoluegnes

java/RPCClient.java

@@ -1,53 +1,57 @@
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Envelope;
+
+import java.io.IOException;
 import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeoutException;
 
 public class RPCClient {
 
   private Connection connection;
   private Channel channel;
   private String requestQueueName = "rpc_queue";
   private String replyQueueName;
-  private QueueingConsumer consumer;
 
-  public RPCClient() throws Exception {
+  public RPCClient() throws IOException, TimeoutException {
     ConnectionFactory factory = new ConnectionFactory();
     factory.setHost("localhost");
+
     connection = factory.newConnection();
     channel = connection.createChannel();
 
     replyQueueName = channel.queueDeclare().getQueue();
-    consumer = new QueueingConsumer(channel);
-    channel.basicConsume(replyQueueName, true, consumer);
   }
 
-  public String call(String message) throws Exception {
-    String response = null;
+  public String call(String message) throws IOException, InterruptedException {
     String corrId = UUID.randomUUID().toString();
 
-    BasicProperties props = new BasicProperties
-                                .Builder()
-                                .correlationId(corrId)
-                                .replyTo(replyQueueName)
-                                .build();
+    AMQP.BasicProperties props = new AMQP.BasicProperties
+            .Builder()
+            .correlationId(corrId)
+            .replyTo(replyQueueName)
+            .build();
 
     channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
 
-    while (true) {
-      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-      if (delivery.getProperties().getCorrelationId().equals(corrId)) {
-        response = new String(delivery.getBody(),"UTF-8");
-        break;
+    final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
+
+    channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
+      @Override
+      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+        response.offer(new String(body, "UTF-8"));
       }
-    }
+    });
 
-    return response;
+    return response.take();
   }
 
-  public void close() throws Exception {
+  public void close() throws IOException {
     connection.close();
   }
 
@@ -61,15 +65,15 @@ public static void main(String[] argv) {
       response = fibonacciRpc.call("30");
       System.out.println(" [.] Got '" + response + "'");
     }
-    catch  (Exception e) {
+    catch  (IOException | TimeoutException | InterruptedException e) {
       e.printStackTrace();
     }
     finally {
       if (fibonacciRpc!= null) {
         try {
           fibonacciRpc.close();
         }
-        catch (Exception ignore) {}
+        catch (IOException _ignore) {}
       }
     }
   }

java/RPCServer.java

@@ -1,78 +1,83 @@
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.AMQP.BasicProperties;
-  
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Envelope;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
 public class RPCServer {
-  
+
   private static final String RPC_QUEUE_NAME = "rpc_queue";
-  
+
   private static int fib(int n) {
     if (n ==0) return 0;
     if (n == 1) return 1;
     return fib(n-1) + fib(n-2);
   }
-    
+
   public static void main(String[] argv) {
+    ConnectionFactory factory = new ConnectionFactory();
+    factory.setHost("localhost");
+
     Connection connection = null;
-    Channel channel = null;
     try {
-      ConnectionFactory factory = new ConnectionFactory();
-      factory.setHost("localhost");
-  
-      connection = factory.newConnection();
-      channel = connection.createChannel();
-      
+      connection      = factory.newConnection();
+      Channel channel = connection.createChannel();
+
       channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
-  
+
       channel.basicQos(1);
-  
-      QueueingConsumer consumer = new QueueingConsumer(channel);
-      channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
-  
+
       System.out.println(" [x] Awaiting RPC requests");
-  
-      while (true) {
-        String response = null;
-        
-        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-        
-        BasicProperties props = delivery.getProperties();
-        BasicProperties replyProps = new BasicProperties
-                                         .Builder()
-                                         .correlationId(props.getCorrelationId())
-                                         .build();
-        
-        try {
-          String message = new String(delivery.getBody(),"UTF-8");
-          int n = Integer.parseInt(message);
-  
-          System.out.println(" [.] fib(" + message + ")");
-          response = "" + fib(n);
-        }
-        catch (Exception e){
-          System.out.println(" [.] " + e.toString());
-          response = "";
-        }
-        finally {  
-          channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes("UTF-8"));
-  
-          channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+
+      Consumer consumer = new DefaultConsumer(channel) {
+        @Override
+        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+          AMQP.BasicProperties replyProps = new AMQP.BasicProperties
+                  .Builder()
+                  .correlationId(properties.getCorrelationId())
+                  .build();
+
+          String response = "";
+
+          try {
+            String message = new String(body,"UTF-8");
+            int n = Integer.parseInt(message);
+
+            System.out.println(" [.] fib(" + message + ")");
+            response += fib(n);
+          }
+          catch (RuntimeException e){
+            System.out.println(" [.] " + e.toString());
+          }
+          finally {
+            channel.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
+
+            channel.basicAck(envelope.getDeliveryTag(), false);
+          }
         }
+      };
+
+      channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
+
+      //loop to prevent reaching finally block
+      while(true) {
+        try {
+          Thread.sleep(100);
+        } catch (InterruptedException _ignore) {}
       }
-    }
-    catch  (Exception e) {
+    } catch (IOException | TimeoutException e) {
       e.printStackTrace();
     }
     finally {
-      if (connection != null) {
+      if (connection != null)
         try {
           connection.close();
-        }
-        catch (Exception ignore) {}
-      }
-    }      		      
+        } catch (IOException _ignore) {}
+    }
   }
-}
-
+}