initial commit

Author: pratapihemant.patel

.gitignore

@@ -0,0 +1,2 @@
+/bin/
+/build/

build.gradle

@@ -0,0 +1,19 @@
+apply plugin: 'java'
+apply plugin: 'eclipse'
+
+jar {
+	baseName = 'rabbitmq-sample'
+	version = '0.0.1-SNAPSHOT'
+}
+
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    compile group: 'com.rabbitmq', name: 'amqp-client', version: '3.6.6'
+    testCompile group: 'junit', name: 'junit', version: '4.+'
+}

src/main/java/com/github/rabbitmq/CommonUtils.java

@@ -0,0 +1,31 @@
+package com.github.rabbitmq;
+
+import java.io.IOException;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+public class CommonUtils {
+
+    public static void close(Connection connection) {
+
+        try {
+            if (null != connection) {
+                connection.close();
+            }
+        }
+        catch (IOException e) {
+        }
+    }
+
+    public static void close(Channel channel) {
+
+        try {
+            if (null != channel) {
+                channel.close();
+            }
+        }
+        catch (Exception e) {
+        }
+    }
+}

src/main/java/com/github/rabbitmq/queue/CommonConstants.java

@@ -0,0 +1,8 @@
+package com.github.rabbitmq.queue;
+
+interface CommonConstants {
+
+	String QUEUE_NAME = "hello";
+	String HOST_NAME = "localhost";
+	
+}

src/main/java/com/github/rabbitmq/queue/QueueReceiver.java

@@ -0,0 +1,77 @@
+package com.github.rabbitmq.queue;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+import com.github.rabbitmq.CommonUtils;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+
+public class QueueReceiver implements AutoCloseable {
+
+    private Connection connection;
+    private Channel channel;
+
+    public void run() {
+
+        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
+
+        try {
+            initConnection();
+
+            receive();
+
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    private void receive() throws IOException {
+
+        Consumer consumer = new DefaultConsumer(channel) {
+
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
+                    byte[] body) throws IOException {
+
+                String message = new String(body, "UTF-8");
+                System.out.println(" [x] Received '" + message + "'");
+            }
+        };
+
+        while (true) {
+            channel.basicConsume(CommonConstants.QUEUE_NAME, true, consumer);
+        }
+    }
+
+    private void initConnection() throws IOException, TimeoutException {
+
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost(CommonConstants.HOST_NAME);
+        connection = factory.newConnection();
+        channel = connection.createChannel();
+        channel.queueDeclare(CommonConstants.QUEUE_NAME, false, false, false, null);
+    }
+
+    @Override
+    public void close() {
+
+        CommonUtils.close(channel);
+        CommonUtils.close(connection);
+        System.out.println("receiver connection closed");
+    }
+
+    public static void main(String[] args) {
+
+        try (QueueReceiver receiver = new QueueReceiver()) {
+            receiver.run();
+        }
+    }
+}

src/main/java/com/github/rabbitmq/queue/QueueSender.java

@@ -0,0 +1,66 @@
+package com.github.rabbitmq.queue;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Scanner;
+import java.util.concurrent.TimeoutException;
+
+import com.github.rabbitmq.CommonUtils;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+public class QueueSender implements Runnable, AutoCloseable {
+
+    private Connection connection;
+    private Channel channel;
+
+    @Override
+    public void run() {
+
+        System.out.println(" [*] Enter messages to send. To exit press CTRL+C");
+        try {
+            initConnection();
+            send();
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void send() throws UnsupportedEncodingException, IOException {
+
+        try (Scanner sc = new Scanner(System.in)) {
+
+            while (sc.hasNextLine()) {
+                String message = sc.nextLine();
+                channel.basicPublish("", CommonConstants.QUEUE_NAME, null, message.getBytes("UTF-8"));
+                System.out.println("Message Sent: " + message);
+            }
+        }
+    }
+
+    private void initConnection() throws IOException, TimeoutException {
+
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost(CommonConstants.HOST_NAME);
+        connection = factory.newConnection();
+        channel = connection.createChannel();
+        channel.queueDeclare(CommonConstants.QUEUE_NAME, false, false, false, null);
+    }
+
+    @Override
+    public void close() {
+
+        CommonUtils.close(channel);
+        CommonUtils.close(connection);
+        System.out.println("sender connection closed");
+    }
+
+    public static void main(String[] args) {
+
+        try (QueueSender sender = new QueueSender()) {
+            sender.run();
+        }
+    }
+}