package com.github.triceo.splitlog;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.triceo.splitlog.conditions.MessageDeliveryCondition;

final class MessageExchange {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageExchange.class);

    private MessageDeliveryCondition messageBlockingCondition = null;
    private final Exchanger<Message> messageExchanger = new Exchanger<Message>();

    public void notifyOfMessage(final Message msg, final MessageDeliveryStatus status,
        final MessageDeliveryNotificationSource source) {
        MessageExchange.LOGGER.info("Follower {} notified of message {} in state {}.", this, msg, status);
        if (this.messageBlockingCondition == null) {
            MessageExchange.LOGGER.debug("Follower {} not waiting for message {} in state {}.", this, msg, status);
            // this does nothing with the message
            return;
        }
        // check if the user code accepts the message
        if (source instanceof LogWatch) {
            if (!this.messageBlockingCondition.accept(msg, status, (LogWatch) source)) {
                return;
            }
        } else if (source instanceof AbstractLogWatchFollower) {
            if (!this.messageBlockingCondition.accept(msg, status, (Follower) source)) {
                return;
            }
        } else {
            throw new IllegalStateException(source + " is not a valid message notification source.");
        }
        MessageExchange.LOGGER.debug("Follower {} accepted message {} in state {}.", this, msg, status);
        this.messageBlockingCondition = null;
        try {
            this.messageExchanger.exchange(msg);
        } catch (final InterruptedException e) {
            MessageExchange.LOGGER
                    .warn("Notifying follower {} of message {} in state {} failed.", this, msg, status, e);
        }
    }

    /*
     * The method is synchronized, therefore no thread will be able to overwrite
     * an already set condition. That condition will later be unset by the
     * notify*() method calls from the tailing thread.
     */
    public synchronized Message waitForMessage(final MessageDeliveryCondition condition, final long timeout,
        final TimeUnit unit) {
        this.messageBlockingCondition = condition;
        try {
            if (timeout < 0) {
                return this.messageExchanger.exchange(null);
            } else {
                return this.messageExchanger.exchange(null, timeout, unit);
            }
        } catch (final TimeoutException e) {
            return null;
        } catch (final InterruptedException e) {
            return null;
        } finally { // just in case
            this.messageBlockingCondition = null;
        }
    }

}

@@ -740,29 +740,17 @@
 R.info(%22
-Follower %7B%7D n
+N
 otified 
@@ -770,32 +770,34 @@
  in state %7B%7D
-.%22, this
+ from %7B%7D.%22
 , msg, statu
@@ -789,32 +789,40 @@
 %7D.%22, msg, status
+, source
 );%0A        if (t
@@ -907,21 +907,9 @@
 ug(%22
-Follower %7B%7D n
+N
 ot w
@@ -937,32 +937,34 @@
  in state %7B%7D
-.%22, this
+ from %7B%7D.%22
 , msg, statu
@@ -956,32 +956,40 @@
 %7D.%22, msg, status
+, source
 );%0A            /
@@ -1641,21 +1641,9 @@
 ug(%22
-Follower %7B%7D a
+A
 ccep
@@ -1668,24 +1668,26 @@
 state %7B%7D
-.%22, this
+ from %7B%7D.%22
 , msg, s
@@ -1691,16 +1691,24 @@
 , status
+, source
 );%0A     
@@ -1898,57 +1898,40 @@
 GGER
-%0A                    .warn(%22Notifying f
+.warn(%22Failed to notify F
 ollower
- %7B%7D
  of 
@@ -1958,34 +1958,57 @@
 %7B%7D f
-ailed.%22, this, msg, status
+rom %7B%7D.%22, msg, status,%0A                    source
 , e)

Fix log messages in MessageExchange