package net.intelie.lognit.cli.input;

import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
import net.intelie.lognit.cli.http.RestListener;
import net.intelie.lognit.cli.model.Message;
import net.intelie.lognit.cli.model.MessageBag;

import java.util.Deque;
import java.util.LinkedList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class BufferListener implements RestListener<MessageBag> {
    public static final String NO_CLUSTER_INFO = "WARN: seems there is a bug in server response, no cluster info";
    public static final String MISSING_NODES_RESPONSE = "WARN: missing some cluster responses, check nodes status";
    public static final String QUERY_CANCELLED = "WARN: query has been cancelled (%s)";
    private final Deque<MessageBag> historic;
    private final Deque<MessageBag> other;
    private final Semaphore semaphore;
    private final MessagePrinter printer;
    private boolean releasing;

    @Inject
    public BufferListener(MessagePrinter printer) {
        this.printer = printer;
        this.historic = new LinkedList<MessageBag>();
        this.other = new LinkedList<MessageBag>();
        this.semaphore = new Semaphore(0);
        this.releasing = false;
    }


    @Override
    public synchronized void receive(MessageBag messages) {
        if (releasing) {
            printBag(messages);
            return;
        }
        if (messages.isHistoric()) {
            historic.add(messages);
            semaphore.release();
        } else {
            other.add(messages);
        }
    }

    public boolean waitHistoric(int timeout, int releaseMax) {
        boolean success = false;
        try {
            if (!semaphore.tryAcquire(1, timeout, TimeUnit.MILLISECONDS)) {
                printer.printStatus(MISSING_NODES_RESPONSE);
                return false;
            }
            int waiting = historic.getFirst().getTotalNodes() - 1;
            if (waiting < 0) {
                printer.printStatus(NO_CLUSTER_INFO);
                Thread.sleep(timeout);
                success = false;
            } else {
                success = semaphore.tryAcquire(waiting, timeout, TimeUnit.MILLISECONDS);
                if (!success) {
                    printer.printStatus(MISSING_NODES_RESPONSE);
                }
            }
        } catch (InterruptedException ex) {
            printer.printStatus(MISSING_NODES_RESPONSE);
        }
        releaseHistoric(releaseMax);
        return success;
    }

    private void releaseHistoric(int releaseMax) {
        List<Message> reverse = pickValidHistory(releaseMax);

        for (Message message : reverse)
            printer.printMessage(message);
    }

    private List<Message> pickValidHistory(int releaseMax) {
        PriorityQueue<Message> queue = new PriorityQueue<Message>();

        while (!historic.isEmpty()) {
            MessageBag bag = historic.pop();
            queue.addAll(bag.getItems());
        }

        Iterable<Message> limited = Iterables.limit(queue, releaseMax);
        LinkedList<Message> list = Lists.newLinkedList(limited);
        return Lists.reverse(list);
    }

    public synchronized void releaseAll() {
        releasing = true;
        while (!other.isEmpty()) {
            MessageBag bag = other.pop();
            printBag(bag);
        }
    }

    private void printBag(MessageBag bag) {
        if (bag.isSuccess())
            for (Message message : Lists.reverse(bag.getItems()))
                printer.printMessage(message);
        else
            printer.printStatus(QUERY_CANCELLED, bag.getMessage());
    }

    public MessagePrinter getPrinter() {
        return printer;
    }
}

@@ -36,52 +36,8 @@
 t;%0A%0A
-import com.google.common.collect.Iterables;%0A
 impo
@@ -2906,17 +2906,16 @@
 age%3E();%0A
-%0A
         
@@ -2933,34 +2933,32 @@
 toric.isEmpty())
- %7B
 %0A            Mes
@@ -2958,152 +2958,160 @@
     
-MessageBag bag = historic.pop();%0A            queue.addAll(bag.getItems());%0A        %7D%0A%0A        Iterable%3CMessage%3E limited = Iterables.limit(queue,
+queue.addAll(historic.pop().getItems());%0A%0A        LinkedList%3CMessage%3E list = new LinkedList%3CMessage%3E();%0A        while (!queue.isEmpty() && list.size() %3C
  rel
@@ -3110,33 +3110,32 @@
 () %3C releaseMax)
-;
 %0A        LinkedL
@@ -3131,65 +3131,42 @@
     
-LinkedList%3CMessage%3E list = Lists.newLinkedList(limited
+    list.addFirst(queue.poll()
 );%0A
+%0A
     
@@ -3180,27 +3180,12 @@
 urn 
-Lists.reverse(
 list
-)
 ;%0A  

fix message sorting