/*
 * ICESOFT COMMERCIAL SOURCE CODE LICENSE V 1.1
 *
 * The contents of this file are subject to the ICEsoft Commercial Source
 * Code License Agreement V1.1 (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.icesoft.com/license/commercial-source-v1.1.html
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations under
 * the License.
 *
 * Copyright 2009-2015 ICEsoft Technologies Canada, Corp. All Rights Reserved.
 */
package com.icesoft.net.messaging;

import com.icesoft.util.Properties;

import java.io.Serializable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

public class QueueMessagePublisher
implements MessagePublisher {
    private static final Logger LOGGER = Logger.getLogger(QueueMessagePublisher.class.getName());

    private static final Runnable NOOP =
        new Runnable() {
            public void run() {
                // do nothing...
            }
        };

    private final LinkedBlockingQueue<Runnable> messageQueue = new LinkedBlockingQueue<Runnable>();

    private final DefaultMessageService defaultMessageService;
    private final MessageServiceClient messageServiceClient;
    private final ScheduledFuture scheduledFuture;

    private boolean stopped = false;

    public QueueMessagePublisher(
        final DefaultMessageService defaultMessageService,
        final MessageServiceClient messageServiceClient,
        final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {

        this.defaultMessageService = defaultMessageService;
        this.messageServiceClient = messageServiceClient;
        this.scheduledFuture =
            scheduledThreadPoolExecutor.schedule(
                new Runnable() {
                    public void run() {
                        while (!stopped) {
                            try {
                                try {
                                    messageQueue.take().run();
                                } catch (InterruptedException exception) {
                                    // todo: what to do?
                                }
                            } catch (final Exception exception) {
                                if (LOGGER.isLoggable(Level.WARNING)) {
                                    LOGGER.log(
                                        Level.WARNING,
                                        "Exception caught on " + this.getClass().getName() + " Runnable.",
                                        exception);
                                }
                            }
                        }
                        messageQueue.clear();
                        scheduledFuture.cancel(false);
                    }
                },
                0,
                TimeUnit.MILLISECONDS);
    }

    public void publish(final Serializable objectMessage) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(objectMessage);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish object message '" + objectMessage + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final Serializable objectMessage, final Properties messageProperties) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(objectMessage, messageProperties);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish object message '" + objectMessage + "' " +
                                        "using message properties '" + messageProperties + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final Serializable objectMessage, final Properties messageProperties, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(objectMessage, messageProperties, messageType);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish object message '" + objectMessage + "' " +
                                        "using message properties '" + messageProperties + "' " +
                                        "with message type '" + messageType + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final Serializable objectMessage, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(objectMessage, messageType);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish object message '" + objectMessage + "' " +
                                        "with message type '" + messageType + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final String textMessage) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(textMessage);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish text message '" + textMessage + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final String textMessage, final Properties messageProperties) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(textMessage, messageProperties);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish text message '" + textMessage + "' " +
                                        "using message properties '" + messageProperties + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final String textMessage, final Properties messageProperties, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(textMessage, messageProperties, messageType);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish text message '" + textMessage + "' " +
                                        "using message properties '" + messageProperties + "' " +
                                        "with message type '" + messageType + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publish(final String textMessage, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publish(textMessage, messageType);
                        } catch (final Exception exception) {
                            throw
                                new MessageServiceException(
                                    "Unable to publish text message '" + textMessage + "' " +
                                        "with message type '" + messageType + "'.",
                                    exception);
                        }
                    }
                });
        }
    }

    public void publishNow(final Serializable objectMessage) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(objectMessage);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final Serializable objectMessage, final Properties messageProperties) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(objectMessage, messageProperties);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final Serializable objectMessage, final Properties messageProperties, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(objectMessage, messageProperties, messageType);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final Serializable objectMessage, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(objectMessage, messageType);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final String textMessage) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publishNow(textMessage);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final String textMessage, final Properties messageProperties) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(textMessage, messageProperties);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final String textMessage, final Properties messageProperties, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            // throws MessageServiceException
                            messageServiceClient.publishNow(textMessage, messageProperties, messageType);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void publishNow(final String textMessage, final String messageType) {
        if (!stopped) {
            messageQueue.offer(
                new Runnable() {
                    public void run() {
                        try {
                            messageServiceClient.publishNow(textMessage, messageType);
                        } catch (MessageServiceException exception) {
                            if (defaultMessageService.reconnectNow()) {
                                run();
                            }
                        }
                    }
                });
        }
    }

    public void stop() {
        stopped = true;
        messageQueue.offer(NOOP);
    }
}