package org.datavaultplatform.worker.queue;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.datavaultplatform.common.task.Task;
import org.datavaultplatform.common.task.Context;

public class Receiver {

    private String queueServer;
    private String queueName;
    private String queueUser;
    private String queuePassword;
    private String archiveDir;
    private String tempDir;
    private String metaDir;

    public void setQueueServer(String queueServer) {
        this.queueServer = queueServer;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public void setQueueUser(String queueUser) {
        this.queueUser = queueUser;
    }

    public void setQueuePassword(String queuePassword) {
        this.queuePassword = queuePassword;
    }

    public void setArchiveDir(String archiveDir) {
        this.archiveDir = archiveDir;
    }

    public void setTempDir(String tempDir) {
        this.tempDir = tempDir;
    }
    
    public void setMetaDir(String metaDir) {
        this.metaDir = metaDir;
    }

    public void receive(EventSender events) throws IOException, InterruptedException, TimeoutException {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(queueServer);
        factory.setUsername(queueUser);
        factory.setPassword(queuePassword);
        
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);
        System.out.println(" [*] Waiting for messages.");
        
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");

            // Decode and begin the job ...
            
            try {
                ObjectMapper mapper = new ObjectMapper();
                Task commonTask = mapper.readValue(message, Task.class);
                
                Class<?> clazz = Class.forName(commonTask.getTaskClass());
                Task concreteTask = (Task)(mapper.readValue(message, clazz));
                
                Context context = new Context(archiveDir, tempDir, metaDir, events);
                concreteTask.performAction(context);
                
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        
        // Unreachable - this demo never terminates
        // channel.close();
        // connection.close();
    }
}
@@ -1864,25 +1864,17 @@
 ges.%22);%0A
-        
 %0A
+
         
@@ -1931,16 +1931,46 @@
 hannel);
+%0A%0A        channel.basicQos(1);
 %0A       
@@ -2002,19 +2002,20 @@
 ueName, 
-tru
+fals
 e, consu
@@ -2873,32 +2873,111 @@
 );%0A            %7D
+%0A%0A            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
 %0A        %7D%0A     

Change to manual acknowledgement of messages