package org.ow2.chameleon.fuchsia.mqtt.importer.amqp;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import org.apache.felix.ipojo.annotations.*;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.ow2.chameleon.fuchsia.core.declaration.ImportDeclaration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Dictionary;
import java.util.Hashtable;

@Component(name = "AMQPJointFactory")
@Provides
public class AMQPOutputRouter implements Runnable {

    @Requires
    EventAdmin eventAdmin;

    private Connection connection;

    private Channel channel;

    private boolean isMonitoringArrivals;

    private Thread eventArrivalMonitor;

    @Property(name = "mqtt.queue")
    String queue;

    @Validate
    public void start() throws IOException {

        ConnectionFactory factory = new ConnectionFactory();
        connection = factory.newConnection();
        channel = connection.createChannel();
        channel.queueDeclare(queue, false, false, false, null);

        eventArrivalMonitor = new Thread(this);
        eventArrivalMonitor.start();

        isMonitoringArrivals =true;

    }

    @Invalidate
    public void stop(){
        isMonitoringArrivals =false;
    }

    public void run(){

        try {

            getLogger().info("Monitoring AMPQ Queue named '{}'", queue);

            QueueingConsumer consumer = new QueueingConsumer(channel);

            channel.basicConsume(queue, true, consumer);

            while (isMonitoringArrivals) {

                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());

                getLogger().info("AMQP: message '{}' received,", message);

                getLogger().info("Forwarding ..");

                eventAdmin.sendEvent(getEventAdminMessage(new Hashtable()));

                getLogger().info("EventAdmin: message '{}' queue '{}' sent", message,queue);

            }

        } catch (Exception e) {

            getLogger().error("Failed to monitor AMPQ Queue named '{}' duo to {}", queue,e.getMessage());

        }

    }

    private Event getEventAdminMessage(Dictionary properties) {

        Event eventAdminMessage = new Event(queue, properties);

        return eventAdminMessage;

    }

    public boolean validate(ImportDeclaration declaration) {
        return false;
    }

    public Logger getLogger() {
        return LoggerFactory.getLogger(this.getClass());
    }

}

@@ -80,145 +80,9 @@
 ent.
-Channel;%0Aimport com.rabbitmq.client.Connection;%0Aimport com.rabbitmq.client.ConnectionFactory;%0Aimport com.rabbitmq.client.QueueingConsumer
+*
 ;%0Aim
@@ -709,16 +709,245 @@
 nitor;%0A%0A
+    @Property(name = %22mqtt.server.host%22,value = ConnectionFactory.DEFAULT_HOST)%0A    private String serverHost;%0A%0A    @Property(name = %22mqtt.server.port%22,value = ConnectionFactory.DEFAULT_AMQP_PORT+%22%22)%0A    private int serverPort;%0A%0A
     @Pro
@@ -1156,16 +1156,65 @@
 nection(
+new Address%5B%5D%7Bnew Address(serverHost,serverPort)%7D
 );%0A     
@@ -2088,24 +2088,129 @@
 ding ..%22);%0A%0A
+                Hashtable metatable=new Hashtable();%0A                metatable.put(%22content%22, message);%0A%0A
             
@@ -2255,31 +2255,25 @@
 Message(
-new Hash
+meta
 table
-()
 ));%0A%0A   

solve issue #12