/*
 * Copyright 2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.dempsy.transport.blockingqueue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

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

import net.dempsy.Infrastructure;
import net.dempsy.transport.Listener;
import net.dempsy.transport.MessageTransportException;
import net.dempsy.transport.NodeAddress;
import net.dempsy.transport.Receiver;
import net.dempsy.util.SafeString;

/**
 * <p>
 * The Message transport default library comes with this BlockingQueue implementation.
 * </p>
 * 
 * <p>
 * This class represents both the MessageTransportSender and the concrete Adaptor (since BlockingQueues don't span process spaces). You need to initialize it with the BlockingQueue to use, as well as the
 * MessageTransportListener to send messages to.
 * </p>
 * 
 * <p>
 * Optionally you can provide it with a name that will be used in the thread that's started to read messages from the queue.
 * </p>
 */
public class BlockingQueueReceiver implements Runnable, Receiver {
    private static Logger LOGGER = LoggerFactory.getLogger(BlockingQueueReceiver.class);

    private final BlockingQueueAddress address;
    private final BlockingQueue<Object> queue;

    private Listener<Object> listener = null;

    private final AtomicBoolean running = new AtomicBoolean(false);
    private Thread currentThread = null;
    private boolean shutdown;

    private final static AtomicLong guidGenerator = new AtomicLong(0);

    /**
     * <p>
     * This method starts a background thread that reads messages from the queue and sends them to a registered MessageTransportListener.
     * </p>
     * 
     * <p>
     * This method is tagged with a @PostConstruct but outside of a dependency injection container that's set up to manage default java lifecycle, it needs to be called explicitly.
     * </p>
     * 
     * <p>
     * Alternatively you can manage the threading yourself since a BlockingQueueAdaptor itself is the Runnable that's started.
     * </p>
     * 
     * @throws MessageTransportException
     *             if the BlockingQueue implementation isn't set or if the listener to send the messages to isn't set.
     */
    public BlockingQueueReceiver(final BlockingQueue<Object> queue) {
        this.queue = queue;
        this.address = new BlockingQueueAddress(queue, "BlockingQueue_" + guidGenerator.getAndIncrement());
    }

    @Override
    public void run() {
        synchronized (this) {
            currentThread = Thread.currentThread();

            if (shutdown == true)
                return;

            running.set(true);
        }

        final Listener<Object> curListener = listener;

        // This check is cheap but unlocked
        while (!shutdown) {
            try {
                final Object val = queue.take();
                curListener.onMessage(val);
            } catch (final InterruptedException ie) {
                synchronized (this) {
                    // if we were interrupted we're probably stopping.
                    if (!shutdown)
                        LOGGER.warn("Superfluous interrupt.", ie);
                }
            } catch (final MessageTransportException err) {
                LOGGER.error("Exception while handling message.", err);
            }
        }

        running.set(false);
    }

    /**
     * A BlockingQueueAdaptor requires a MessageTransportListener to be set in order to adapt a client side.
     * 
     * @param listener
     *            is the MessageTransportListener to push messages to when they come in.
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    @Override
    public synchronized void start(final Listener listener, final Infrastructure infra) {
        if (listener == null)
            throw new IllegalArgumentException("Cannot pass null to " + BlockingQueueReceiver.class.getSimpleName() + ".setListener");
        if (this.listener != null)
            throw new IllegalStateException(
                    "Cannot set a new Listener (" + SafeString.objectDescription(listener) + ") on a " + BlockingQueueReceiver.class.getSimpleName()
                            + " when there's one already set (" + SafeString.objectDescription(this.listener) + ")");
        this.listener = listener;
        infra.getThreadingModel().runDaemon(this, "BQReceiver-" + address.toString());
    }

    @Override
    public void close() {
        synchronized (this) {
            shutdown = true;
        }

        while (running.get()) {
            if (currentThread != null)
                currentThread.interrupt();
            Thread.yield();
        }

        address.close();
    }

    @Override
    public NodeAddress getAddress(final Infrastructure infra) {
        return this.address;
    }

}

@@ -232,21 +232,16 @@
 at%0A *%0A *
-     
  http://
@@ -1191,33 +1191,32 @@
 tion.%0A * %3C/p%3E%0A *
- 
 %0A * %3Cp%3E%0A * This 
@@ -1360,24 +1360,27 @@
 nitialize it
+%0A *
  with the Bl
@@ -1473,17 +1473,16 @@
  %3C/p%3E%0A *
- 
 %0A * %3Cp%3E%0A
@@ -2134,723 +2134,8 @@
 );%0A%0A
-    /**%0A     * %3Cp%3E%0A     * This method starts a background thread that reads messages from the queue and sends them to a registered MessageTransportListener.%0A     * %3C/p%3E%0A     * %0A     * %3Cp%3E%0A     * This method is tagged with a @PostConstruct but outside of a dependency injection container that's set up to manage default java lifecycle, it needs to be called explicitly.%0A     * %3C/p%3E%0A     * %0A     * %3Cp%3E%0A     * Alternatively you can manage the threading yourself since a BlockingQueueAdaptor itself is the Runnable that's started.%0A     * %3C/p%3E%0A     * %0A     * @throws MessageTransportException%0A     *             if the BlockingQueue implementation isn't set or if the listener to send the messages to isn't set.%0A     */%0A
     
@@ -2393,33 +2393,32 @@
     synchronized
- 
 (this) %7B%0A       
@@ -2477,17 +2477,16 @@
       if
- 
 (shutdow
@@ -2676,17 +2676,16 @@
    while
- 
 (!shutdo
@@ -2816,25 +2816,24 @@
      %7D catch
- 
 (final Inter
@@ -2874,33 +2874,32 @@
     synchronized
- 
 (this) %7B%0A       
@@ -2984,17 +2984,16 @@
       if
- 
 (!shutdo
@@ -3100,17 +3100,16 @@
  %7D catch
- 
 (final M
@@ -3395,17 +3395,16 @@
 .%0A     *
- 
 %0A     * 
@@ -3425,23 +3425,16 @@
 r%0A     *
-       
      is 
@@ -3532,17 +3532,16 @@
 rnings(%7B
- 
 %22rawtype
@@ -3543,17 +3543,16 @@
 wtypes%22,
- 
 %22uncheck
@@ -3554,17 +3554,16 @@
 checked%22
- 
 %7D)%0A    @
@@ -3671,17 +3671,16 @@
       if
- 
 (listene
@@ -3835,17 +3835,16 @@
       if
- 
 (this.li
@@ -3920,20 +3920,16 @@
         
-    
 %22Cannot 
@@ -4049,24 +4049,16 @@
 eName()%0A
-        
         
@@ -4347,17 +4347,16 @@
 hronized
- 
 (this) %7B
@@ -4409,17 +4409,16 @@
    while
- 
 (running
@@ -4441,17 +4441,16 @@
       if
- 
 (current

remove incorrect javadoc.