package org.rabix.transport.mechanism.impl.rabbitmq;

import java.io.IOException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;

import org.apache.commons.configuration.Configuration;
import org.rabix.common.json.BeanSerializer;
import org.rabix.common.json.processor.BeanProcessorException;
import org.rabix.transport.mechanism.TransportPlugin;
import org.rabix.transport.mechanism.TransportPluginException;
import org.rabix.transport.mechanism.TransportPluginType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class TransportPluginRabbitMQ implements TransportPlugin<TransportQueueRabbitMQ> {

  public static final String DEFAULT_ENCODING = "UTF-8";

  private static final Logger logger = LoggerFactory.getLogger(TransportPluginRabbitMQ.class);
  
  private ConnectionFactory factory;
  
  public TransportPluginRabbitMQ(Configuration configuration) throws TransportPluginException {
    factory = new ConnectionFactory();
    
    if (TransportConfigRabbitMQ.isDev(configuration)) {
      factory.setHost("localhost");
    } else {
      factory.setHost(TransportConfigRabbitMQ.getHost(configuration));
      factory.setPort(TransportConfigRabbitMQ.getPort(configuration));
      factory.setUsername(TransportConfigRabbitMQ.getUsername(configuration));
      factory.setPassword(TransportConfigRabbitMQ.getPassword(configuration));
      factory.setVirtualHost(TransportConfigRabbitMQ.getVirtualhost(configuration));
      if (TransportConfigRabbitMQ.isSSL(configuration)) {
        try {
          factory.useSslProtocol();
        } catch (KeyManagementException | NoSuchAlgorithmException e) {
          logger.error("Failed to initialize TransportPluginRabbitMQ", e);
          throw new TransportPluginException("Failed to initialize TransportPluginRabbitMQ", e);
        }
      }
    }
  }
  
  @Override
  public <T> ResultPair<T> send(TransportQueueRabbitMQ queue, T entity) {
    Channel channel = null;
    Connection connection = null;
    try {
      connection = factory.newConnection();
      channel = connection.createChannel();
      
      channel.exchangeDeclare(queue.getExchange(), queue.getExchangeType());
      String payload = BeanSerializer.serializeFull(entity);
      channel.basicPublish(queue.getExchange(), queue.getRoutingKey(), null, payload.getBytes(DEFAULT_ENCODING));
      return ResultPair.success();
    } catch (IOException | TimeoutException e) {
      logger.error("Failed to send a message to " + queue, e);
      return ResultPair.fail("Failed to send a message to " + queue, e);
    } finally {
      if (channel != null) {
        try {
          channel.close();
        } catch (Exception ignore) { }
      }
      if (connection != null) {
        try {
          connection.close();
        } catch (Exception ignore) { }
      }
    }
  }

  @Override
  public <T> ResultPair<T> receive(final TransportQueueRabbitMQ queue, final Class<T> clazz, final ReceiveCallback<T> receiveCallback) {
    Channel channel = null;
    Connection connection = null;
    try {
      connection = factory.newConnection();
      channel = connection.createChannel();
      
      channel.exchangeDeclare(queue.getExchange(), queue.getExchangeType());
      String queueName = channel.queueDeclare().getQueue();
      channel.queueBind(queueName, queue.getExchange(), queue.getRoutingKey());

      QueueingConsumer consumer = new QueueingConsumer(channel);
      channel.basicConsume(queueName, true, consumer);

      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      receiveCallback.handleReceive(BeanSerializer.deserialize(message, clazz));
      return ResultPair.<T>success();
    } catch (BeanProcessorException e) {
      logger.error("Failed to deserialize message payload", e);
      return ResultPair.<T> fail("Failed to deserialize message payload", e);
    } catch (TransportPluginException e) {
      logger.error("Failed to handle receive", e);
      return ResultPair.<T> fail("Failed to handle receive", e);
    } catch (Exception e) {
      logger.error("Failed to receive a message from " + queue, e);
      return ResultPair.<T> fail("Failed to receive a message from " + queue, e);
    } finally {
      if (channel != null) {
        try {
          channel.close();
        } catch (Exception ignore) { }
      }
      if (connection != null) {
        try {
          connection.close();
        } catch (Exception ignore) { }
      }
    }
  }
  
  @Override
  public TransportPluginType getType() {
    return TransportPluginType.RABBIT_MQ;
  }

}

@@ -78,146 +78,8 @@
 ion;
-%0Aimport java.security.KeyManagementException;%0Aimport java.security.NoSuchAlgorithmException;%0Aimport java.util.concurrent.TimeoutException;
 %0A%0Aim
@@ -886,16 +886,49 @@
 ss);%0A  %0A
+  private Connection connection;%0A
   privat
@@ -1094,21 +1094,29 @@
 tory();%0A
+%0A
     
-%0A
+try %7B%0A  
     if (
@@ -1161,32 +1161,34 @@
 ation)) %7B%0A      
+  
 factory.setHost(
@@ -1205,16 +1205,18 @@
 %22);%0A    
+  
 %7D else %7B
@@ -1218,24 +1218,26 @@
 lse %7B%0A      
+ 
+ 
 factory.setH
@@ -1281,32 +1281,34 @@
 onfiguration));%0A
+  
       factory.se
@@ -1360,32 +1360,34 @@
 ration));%0A      
+ 
+ 
 factory.setUsern
@@ -1435,32 +1435,34 @@
 onfiguration));%0A
+  
       factory.se
@@ -1526,24 +1526,26 @@
 on));%0A      
+  
 factory.setV
@@ -1613,24 +1613,26 @@
 on));%0A      
+  
 if (Transpor
@@ -1667,38 +1667,24 @@
 uration)) %7B%0A
-        try %7B%0A
           fa
@@ -1720,150 +1720,89 @@
    %7D
- catch (KeyManagementException %7C NoSuchAlgorithmException e) %7B%0A          logger.error(%22Failed to initialize TransportPluginRabbitMQ%22, e);%0A    
+%0A      %7D%0A      connection = factory.newConnection();%0A    %7D catch (Exception e) %7B%0A
     
@@ -1890,34 +1890,16 @@
 Q%22, e);%0A
-        %7D%0A      %7D%0A
     %7D%0A  
@@ -2025,91 +2025,13 @@
     
-Connection connection = null;%0A    try %7B%0A      connection = factory.newConnection();
+try %7B
 %0A   
@@ -2394,27 +2394,8 @@
 ion 
-%7C TimeoutException 
 e) %7B
@@ -2668,131 +2668,8 @@
   %7D%0A
-      if (connection != null) %7B%0A        try %7B%0A          connection.close();%0A        %7D catch (Exception ignore) %7B %7D%0A      %7D%0A
     
@@ -2860,91 +2860,13 @@
     
-Connection connection = null;%0A    try %7B%0A      connection = factory.newConnection();
+try %7B
 %0A   
@@ -4151,131 +4151,8 @@
   %7D%0A
-      if (connection != null) %7B%0A        try %7B%0A          connection.close();%0A        %7D catch (Exception ignore) %7B %7D%0A      %7D%0A
     

refactor rabbitmq transport plugin