Communication channel transparency

src/main/java/org/hobbit/core/Constants.java

@@ -30,6 +30,7 @@ private Constants() {
     }
 
     // =============== ENVIRONMENT CONSTANTS ===============
+    public static final String IS_RABBIT_MQ_ENABLED = "IS_RABBIT_MQ_ENABLED";
 
     public static final String HOBBIT_SESSION_ID_KEY = "HOBBIT_SESSION_ID";
 

src/main/java/org/hobbit/core/com/Channel.java

@@ -0,0 +1,73 @@
+package org.hobbit.core.com;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import org.hobbit.core.components.AbstractCommandReceivingComponent;
+
+import com.rabbitmq.client.AMQP.BasicProperties;
+import org.hobbit.core.rabbit.RabbitMQChannel;
+import org.hobbit.core.com.java.DirectChannel;
+
+/**
+ * An interface with several methods that is implemented by {@link RabbitMQChannel} and {@link DirectChannel}.
+ *
+ * @author Altafhusen Makandar
+ * @author Sourabh Poddar
+ * @author Yamini Punetha
+ * @author Melissa Das
+ *
+ */
+public interface Channel {
+
+    /**
+     * Method to accept messages from the channel.
+     */
+    public void readBytes(Object consumerCallback, Object classs, Boolean autoAck, String queue) throws IOException;
+
+    /**
+     * Method to publish a message of type Byte array the channel.
+     */
+    public void writeBytes(byte data[], String exchange, String routingKey, BasicProperties props) throws IOException;
+
+    /**
+     * Method to publish a message of type ByteBuffer to the channel.
+     */
+    public void writeBytes(ByteBuffer buffer, String exchange, String routingKey, BasicProperties props) throws IOException;
+
+    /**
+     * Method to close a channel.
+     */
+    public void close();
+
+    /**
+     * Method to create a new channel.
+     */
+    public void createChannel() throws Exception;
+
+    /**
+     * Method to get the queue name.
+     */
+    public String getQueueName(AbstractCommandReceivingComponent abstractCommandReceivingComponent)  throws Exception;
+
+    /**
+     * Method to create an exchange for the data transfer for a broker
+     */
+    public void exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
+        Map<String, Object> arguments) throws IOException;
+
+    /**
+     * Method to bind a queue with an exchange. This allows an exchange to publish the messages 
+     * to the queues that are bound to this exchange 
+     */
+    public void queueBind(String queue, String exchange, String routingKey) throws IOException;
+
+    /**
+     * Method to return the channel instance.
+     */
+    public Object getChannel();
+
+    public String declareQueue(String queueName) throws IOException;
+
+}

src/main/java/org/hobbit/core/rabbit/DataHandler.java → src/main/java/org/hobbit/core/com/DataHandler.java

@@ -1,4 +1,4 @@
-package org.hobbit.core.rabbit;
+package org.hobbit.core.com;
 
 public interface DataHandler {
 

src/main/java/org/hobbit/core/rabbit/DataReceiver.java → src/main/java/org/hobbit/core/com/DataReceiver.java

@@ -1,4 +1,4 @@
-package org.hobbit.core.rabbit;
+package org.hobbit.core.com;
 
 import java.io.Closeable;
 

src/main/java/org/hobbit/core/rabbit/DataSender.java → src/main/java/org/hobbit/core/com/DataSender.java

@@ -1,4 +1,4 @@
-package org.hobbit.core.rabbit;
+package org.hobbit.core.com;
 
 import java.io.Closeable;
 import java.io.IOException;

src/main/java/org/hobbit/core/com/java/DirectCallback.java

@@ -0,0 +1,31 @@
+package org.hobbit.core.com.java;
+
+import java.util.List;
+
+import org.hobbit.core.com.Channel;
+
+import com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * This class is used by the DirectChannel implementation 
+ * for a callback function as a consumer callback
+ * @author altaf, sourabh, yamini, melisa
+ *
+ */
+public abstract class  DirectCallback {
+
+    protected Channel channel;
+    protected String queue;
+    protected BasicProperties props;
+
+    public DirectCallback() {}
+
+    public DirectCallback(Channel channel, String queue, BasicProperties props) {
+        this.channel = channel;
+        this.queue = queue;
+        this.props = props;
+    }
+
+    public abstract void callback(byte[] data, List<Object> classs,BasicProperties props);
+
+}

src/main/java/org/hobbit/core/com/java/DirectChannel.java

@@ -0,0 +1,117 @@
+package org.hobbit.core.com.java;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Pipe;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.lang3.StringUtils;
+import org.hobbit.core.com.Channel;
+import org.hobbit.core.components.AbstractCommandReceivingComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.rabbitmq.client.AMQP.BasicProperties;
+/**
+ * This class implements the necessary functionality for message sharing 
+ * without using RabbitMQ. DirectChannel uses Java NIO Pipe for message queuing
+ * implementation
+ *
+ * @author Altafhusen Makandar
+ * @author Sourabh Poddar
+ * @author Yamini Punetha
+ * @author Melissa Das
+ *
+ */
+public class DirectChannel implements Channel {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DirectChannel.class);
+
+    static Map<String, PipeChannel> pipes = new HashMap<>();
+
+    PipeChannel pipeChannel;
+    private ExecutorService threadPool = Executors.newCachedThreadPool();
+    public DirectChannel(){}
+    public DirectChannel(String queue){
+        try {
+            if(pipes.get(queue) == null) {
+                pipeChannel = new PipeChannel(Pipe.open());
+                pipes.put(queue, pipeChannel);
+            }
+        } catch (IOException e) {
+            LOGGER.error("Error creating pipe ",e);
+        }
+    }
+
+
+    @Override
+    public void readBytes(Object callback, Object classs, Boolean autoAck, String queue) {
+        threadPool.execute(new ReadByteChannel(pipes.get(queue), callback, classs));
+    }
+
+    @Override
+    public void writeBytes(byte[] data, String exchange, String routingKey, BasicProperties props) throws IOException {
+        ByteBuffer buffer = ByteBuffer.allocate(data.length);
+        buffer.put(data);
+        writeBytes(buffer, exchange, routingKey, props);
+    }
+
+    @Override
+    public void writeBytes(ByteBuffer buffer, String exchange, String routingKey, BasicProperties props) throws IOException {
+        String queue = StringUtils.isEmpty(exchange) ? routingKey : exchange;
+        if(!pipes.isEmpty()) {
+            pipes.get(queue).setProps(props);
+            String replyQueue = queue;
+            if(props != null && StringUtils.isNotBlank(props.getReplyTo())) {
+                replyQueue = props.getReplyTo();
+            }
+            if(pipes.get(replyQueue).getPipe().sink().isOpen()) {
+                buffer.flip();
+                while (buffer.hasRemaining())
+                    pipes.get(replyQueue).getPipe().sink().write(buffer);
+                buffer.clear();
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+        /*if(ReadByteChannel.classes != null && ReadByteChannel.classes.size() > 0) {
+            ReadByteChannel.classes.clear();
+        }
+        pipes.clear();*/
+    }
+
+    @Override
+    public void createChannel() throws Exception {
+
+    }
+
+    @Override
+    public String getQueueName(AbstractCommandReceivingComponent abstractCommandReceivingComponent) throws Exception {
+        return null;
+    }
+
+    @Override
+    public void exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
+    	    Map<String, Object> arguments) throws IOException {
+
+    }
+
+    @Override
+    public void queueBind(String queue, String exchange, String routingKey) throws IOException {
+
+    }
+
+    @Override
+    public Object getChannel() {
+        return null;
+    }
+
+    @Override
+    public String declareQueue(String queueName) throws IOException {
+        return queueName;
+    }
+}

src/main/java/org/hobbit/core/com/java/DirectReceiverImpl.java

@@ -0,0 +1,65 @@
+package org.hobbit.core.com.java;
+
+import java.io.IOException;
+
+import org.hobbit.core.com.Channel;
+import org.hobbit.core.com.DataHandler;
+import org.hobbit.core.com.DataReceiver;
+import org.hobbit.core.components.communicationfactory.ChannelFactory;
+import org.hobbit.core.data.RabbitQueue;
+
+/**
+ * This class implements the methods of DataReceiver for {@link DirectChannel}
+ *
+ * @author Altafhusen Makander
+ * @author Sourabh Poddar
+ * @author Yamini Punetha
+ * @author Melissa Das
+ *
+ */
+public class DirectReceiverImpl implements DataReceiver {
+
+    public DirectReceiverImpl(String queue, Object consumer) {
+
+        Channel channel = new ChannelFactory().getChannel(false, queue, null);
+        try {
+            channel.readBytes(consumer, this, null, queue);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public int getErrorCount() {
+        return 0;
+    }
+
+    @Override
+    public void closeWhenFinished() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void increaseErrorCount() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public DataHandler getDataHandler() {
+        return null;
+    }
+
+    @Override
+    public RabbitQueue getQueue() {
+        return null;
+    }
+
+}

src/main/java/org/hobbit/core/com/java/DirectSenderImpl.java

@@ -0,0 +1,60 @@
+package org.hobbit.core.com.java;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.hobbit.core.com.Channel;
+import org.hobbit.core.com.DataSender;
+import org.hobbit.core.components.communicationfactory.ChannelFactory;
+/**
+ * DataSender implementation for DirectChannel
+ * @author altaf, sourabh, yamini, melisa
+ *
+ */
+public class DirectSenderImpl implements DataSender {
+
+    Channel senderChannel;
+    String queue;
+
+    public DirectSenderImpl(String queue){
+        this.queue = queue;
+        senderChannel = new ChannelFactory().getChannel(false, queue, null);
+    }
+
+
+    @Override
+    public void close() throws IOException {
+        // TODO Auto-generated method stub
+    }
+
+    @Override
+    public void sendData(byte[] data) throws IOException {
+        ByteBuffer buffer = ByteBuffer.allocate(data.length);
+        buffer.put(data);
+        sleep(100000);
+        senderChannel.writeBytes(buffer, null, this.queue, null);
+        /*try {
+            Thread.sleep(0, 100);
+            sleep(100000);
+            senderChannel.writeBytes(buffer, null, this.queue, null);
+        } catch (Exception e) {
+            LOGGER.error("Error waiting during send data", e);
+        }*/
+    }
+
+    private void sleep(long interval) {
+        long start = System.nanoTime();
+        long totalTime = start + interval;
+        long end = 0;
+        do {
+            end = System.nanoTime();
+        }while(totalTime >= end);
+    }
+
+
+    @Override
+    public void closeWhenFinished() {
+        // TODO Auto-generated method stub
+    }
+
+}