package poussecafe.messaging.internal;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import poussecafe.messaging.MessageAdapter;
import poussecafe.messaging.MessageReceiver;
import poussecafe.messaging.MessageSender;
import poussecafe.runtime.MessageConsumer;
import poussecafe.runtime.OriginalAndMarshaledMessage;

public class InternalMessagingQueue {

    InternalMessagingQueue(MessageConsumer messageConsumer) {
        messageReceiver = new InternalMessageReceiver(messageAdapter, messageConsumer);
        messageSender = new InternalMessageSender(messageAdapter);
    }

    private MessageAdapter messageAdapter = new SerializingMessageAdapter();

    public class InternalMessageReceiver extends MessageReceiver {

        private InternalMessageReceiver(MessageAdapter messageAdapter, MessageConsumer messageConsumer) {
            super(messageAdapter, messageConsumer);
        }

        @Override
        protected void actuallyStartReceiving() {
            receptionThread = new Thread(() -> {
                while (true) {
                    try {
                        Object takenObject = queue.take();
                        mutex.acquire();
                        if (STOP.equals(takenObject)) {
                            return;
                        } else {
                            onMessage(takenObject);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        interrupted = true;
                        return;
                    } finally {
                        // Catch-all, thread must continue to run until explicitly stopped
                        mutex.release();
                    }
                }
            });
            receptionThread.setDaemon(true);
            receptionThread.setName("internal message queue");
            receptionThread.start();
        }

        private Thread receptionThread;

        private boolean interrupted;

        @Override
        protected void actuallyStopReceiving() {
            queue.add(STOP);
        }

        private static final String STOP = "stop";

        public InternalMessagingQueue queue() {
            return InternalMessagingQueue.this;
        }

        @Override
        protected void actuallyInterruptReception() {
            receptionThread.interrupt();
            mutex.release();
        }

        public boolean isInterrupted() {
            return interrupted;
        }
    }

    private Semaphore mutex = new Semaphore(1);

    private BlockingQueue<Object> queue = new LinkedBlockingQueue<>();

    private InternalMessageReceiver messageReceiver;

    public MessageReceiver messageReceiver() {
        return messageReceiver;
    }

    public class InternalMessageSender extends MessageSender {

        private InternalMessageSender(MessageAdapter messageAdapter) {
            super(messageAdapter);
        }

        @Override
        protected void sendMarshalledMessage(OriginalAndMarshaledMessage marshalledMessage) {
            queue.add(marshalledMessage.marshaled());
        }
    }

    private InternalMessageSender messageSender;

    public MessageSender messageSender() {
        return messageSender;
    }

    public void waitUntilEmptyOrInterrupted()
            throws InterruptedException {
        boolean messagesQueued;
        do {
            mutex.acquire();
            if(messageReceiver.isInterrupted()) {
                mutex.release();
                return;
            }
            messagesQueued = !queue.isEmpty();
            mutex.release();
        } while (messagesQueued);
    }
}

@@ -54,27 +54,8 @@
 til.
-concurrent.Blocking
 Queu
@@ -1156,38 +1156,24 @@
     
-Object takenObject = queue.tak
+available.acquir
 e();
@@ -1206,32 +1206,92 @@
 utex.acquire();%0A
+                        Object polledObject = queue.poll();%0A
                 
@@ -1314,21 +1314,22 @@
 .equals(
-taken
+polled
 Object))
@@ -1442,13 +1442,14 @@
 age(
-taken
+polled
 Obje
@@ -2221,16 +2221,49 @@
 (STOP);%0A
+            available.release();%0A
         
@@ -2659,32 +2659,85 @@
        %7D%0A    %7D%0A%0A
+    private Semaphore available = new Semaphore(0);%0A%0A
     private Sema
@@ -2781,24 +2781,16 @@
 private 
-Blocking
 Queue%3COb
@@ -3320,16 +3320,49 @@
 led());%0A
+            available.release();%0A
         

Revert "Remove useless semaphore in internal messaging queue."