package com.xing.beetle;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionException;

/**
 *
 */
public class BeetleConsumer extends DefaultConsumer {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    private final Client client;
    protected final Channel subscriberChannel;
    protected final MessageHandler handler;

    public BeetleConsumer(Client client, Channel subscriberChannel, MessageHandler handler) {
        super(subscriberChannel);

        this.client = client;
        this.subscriberChannel = subscriberChannel;
        this.handler = handler;
    }

    @Override
    public void handleDelivery(String consumerTag, final Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

        final String messageId = properties.getMessageId();
        final String expires_at = properties.getHeaders().get("expires_at").toString();
        Long ttl = 0L;
        if (expires_at != null) {
            ttl = Long.valueOf(expires_at);
        }
        // check for outdated,expired message
        if ((System.currentTimeMillis() / 1000L) > ttl) {
            log.warn("NACK expired message id {} on {}", messageId, subscriberChannel);
            nackSafely(envelope.getDeliveryTag(), false);
            return;
        }

        if (! client.shouldProcessMessage(subscriberChannel, envelope.getDeliveryTag(), messageId)) {
            return;
        }

        final Callable<HandlerResponse> handlerProcessor = handler.process(envelope, properties, body);

        try {
            final Runnable handlerTask = new Runnable() {
                @Override
                public void run() {
                    try {
                        HandlerResponse response = handlerProcessor.call();
                        if (response.isSuccess()) {
                            // TODO move ack to this class instead?
                            client.markMessageAsCompleted(subscriberChannel, envelope.getDeliveryTag(), messageId);
                        } else {
                            // cannot happen right now. delete?
                            nackSafely(envelope.getDeliveryTag(), true);
                        }
                    } catch (Exception e) {
                        final long exceptions = client.incrementExceptions(messageId);
                        final long attempts = client.getAttemptsCount(messageId);
                        // TODO read exceptions/attempts limit from declared Message!
                        if (exceptions > 1 || attempts > 1) {
                            // exceeded configured exception count
                            log.warn("NACK message attempt or exception count exceeded. {} of {} attempts, {} of {} exceptions",
                                new long[] {attempts, 1, exceptions, 1});
                            nackSafely(envelope.getDeliveryTag(), false);
                        } else {
                            client.removeMessageHandlerLock(messageId);
                            nackSafely(envelope.getDeliveryTag(), true);
                        }
                    }
                }
            };

            client.submit(handlerTask);

        } catch (RejectedExecutionException e) {
            log.error("Could not submit message processor to executor! Requeueing message.", e);
            nackSafely(envelope.getDeliveryTag(), true);
        }
    }

    private void nackSafely(long deliveryTag, boolean requeue) {
        try {
            subscriberChannel.basicNack(deliveryTag, false, requeue);
        } catch (IOException e) {
            log.error("Could not NACK message.", e);
        }
    }
}

@@ -901,16 +901,76 @@
 eption %7B
+%0A        final long deliveryTag = envelope.getDeliveryTag();
 %0A%0A      
@@ -1430,32 +1430,24 @@
     
-nackSafely(envelope.getD
+discardMessage(d
 eliv
@@ -1448,33 +1448,24 @@
 (deliveryTag
-(), false
 );%0A         
@@ -1543,37 +1543,25 @@
 berChannel, 
-envelope.getD
+d
 eliveryTag()
@@ -1550,34 +1550,32 @@
 nel, deliveryTag
-()
 , messageId)) %7B%0A
@@ -2143,29 +2143,17 @@
 hannel, 
-envelope.getD
+d
 eliveryT
@@ -2154,18 +2154,16 @@
 iveryTag
-()
 , messag
@@ -2297,32 +2297,34 @@
     
-nackSafely(envelope.getD
+processMessageAgainLater(d
 eliv
@@ -2325,32 +2325,24 @@
 (deliveryTag
-(), true
 );%0A         
@@ -3021,32 +3021,24 @@
     
-nackSafely(envelope.getD
+discardMessage(d
 eliv
@@ -3047,17 +3047,8 @@
 yTag
-(), false
 );%0A 
@@ -3183,32 +3183,34 @@
     
-nackSafely(envelope.getD
+processMessageAgainLater(d
 eliv
@@ -3211,32 +3211,24 @@
 (deliveryTag
-(), true
 );%0A         
@@ -3503,62 +3503,263 @@
     
-nackSafely(envelope.getDeliveryTag(), true);%0A        %7D
+processMessageAgainLater(deliveryTag);%0A        %7D%0A    %7D%0A%0A    private void discardMessage(long deliveryTag) %7B%0A        nackSafely(deliveryTag, false);%0A    %7D%0A%0A    private void processMessageAgainLater(long deliveryTag) %7B%0A        nackSafely(deliveryTag, true);
 %0A   

refactor code to make it a little more readable