package poussecafe.messaging.internal;

import java.util.Queue;
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 {
                        available.acquire();
                        mutex.acquire();
                        Object polledObject = queue.poll();
                        if (STOP.equals(polledObject)) {
                            return;
                        } else {
                            onMessage(polledObject);
                        }
                    } 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);
            available.release();
        }

        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 available = new Semaphore(0);

    private Semaphore mutex = new Semaphore(1);

    private Queue<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());
            available.release();
        }
    }

    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);
    }
}

@@ -50,16 +50,35 @@
 va.util.
+concurrent.Blocking
 Queue;%0Ai
@@ -1175,24 +1175,38 @@
     
-available.acquir
+Object takenObject = queue.tak
 e();
@@ -1251,68 +1251,8 @@
 ();%0A
-                        Object polledObject = queue.poll();%0A
     
@@ -1287,22 +1287,21 @@
 .equals(
-polled
+taken
 Object))
@@ -1414,14 +1414,13 @@
 age(
-polled
+taken
 Obje
@@ -2196,41 +2196,8 @@
 P);%0A
-            available.release();%0A
     
@@ -2609,61 +2609,8 @@
  %7D%0A%0A
-    private Semaphore available = new Semaphore(0);%0A%0A
     
@@ -2666,16 +2666,24 @@
 private 
+Blocking
 Queue%3COb
@@ -3217,41 +3217,8 @@
 ));%0A
-            available.release();%0A
     

Remove useless semaphore in internal messaging queue.