/*
 * 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.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.icepush.Configuration;

public class DefaultMessageService
implements MessageServiceClient.Administrator {
    private static final Logger LOGGER = Logger.getLogger(DefaultMessageService.class.getName());

    private static final int STATE_UNINITIALIZED = 0;
    private static final int STATE_SET_UP = 1;
    private static final int STATE_SET_UP_DONE = 2;
    private static final int STATE_STARTED = 3;
    private static final int STATE_STOPPED = 4;
    private static final int STATE_TEAR_DOWN = 5;
    private static final int STATE_TEAR_DOWN_DONE = 6;
    private static final int STATE_CLOSED = 7;

    private static final int DEFAULT_INTERVAL = 5000;
    private static final int DEFAULT_MAX_RETRIES = 6;
    private static final int DEFAULT_INTERVAL_ON_RECONNECT = 5000;
    private static final int DEFAULT_MAX_RETRIES_ON_RECONNECT = 60;

    private final Object reconnectLock = new Object();
    private final Object stateLock = new Object();

    private final Configuration configuration;
    private final MessageServiceClient messageServiceClient;
    private final boolean retryOnFail;
    private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    private MessagePublisher currentMessagePublisher;
    private long successTimestamp;

    private int currentState = STATE_UNINITIALIZED;
    private int requestedState = STATE_UNINITIALIZED;

    public DefaultMessageService(
        final MessageServiceClient messageServiceClient, final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor)
    throws IllegalArgumentException {
        // throws IllegalArgumentException 
        this(messageServiceClient, scheduledThreadPoolExecutor, null, false);
    }

    public DefaultMessageService(
        final MessageServiceClient messageServiceClient, final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor,
        final boolean retryOnFail)
    throws IllegalArgumentException {
        // throws IllegalArgumentException
        this(messageServiceClient, scheduledThreadPoolExecutor, null, retryOnFail);
    }

    public DefaultMessageService(
        final MessageServiceClient messageServiceClient, final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor,
        final Configuration configuration)
    throws IllegalArgumentException {
        // throws IllegalArgumentException
        this(messageServiceClient, scheduledThreadPoolExecutor, configuration, false);
    }

    public DefaultMessageService(
        final MessageServiceClient messageServiceClient, final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor,
        final Configuration configuration, final boolean retryOnFail)
    throws IllegalArgumentException {

        DefaultMessageServiceContextListener.setMessageService(this);

        if (messageServiceClient == null) {
            throw new IllegalArgumentException("messageServiceClient is null");
        }
        if (scheduledThreadPoolExecutor == null) {
            throw new IllegalArgumentException("scheduledThreadPoolExecutor is null");
        }
        this.messageServiceClient = messageServiceClient;
        this.messageServiceClient.setAdministrator(this);
        this.scheduledThreadPoolExecutor = scheduledThreadPoolExecutor;
        this.configuration = configuration;
        this.retryOnFail = retryOnFail;
        this.currentMessagePublisher =
            new QueueMessagePublisher(this, messageServiceClient, scheduledThreadPoolExecutor);
    }

    public final void close() {
        synchronized (stateLock) {
            requestedState = STATE_CLOSED;
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(Level.FINE, "Requested State: CLOSED");
            }
            if (currentState == STATE_TEAR_DOWN_DONE) {
                try {
                    currentMessagePublisher.stop();
                    // switch from a queue-based message publisher to a noop message publisher.
                    currentMessagePublisher = new NoopMessagePublisher();
                    // throws MessageServiceException
                    messageServiceClient.close();
                    currentState = STATE_CLOSED;
                    if (LOGGER.isLoggable(Level.FINE)) {
                        LOGGER.log(Level.FINE, "Current State: CLOSED");
                    }
                } catch (MessageServiceException exception) {
                    if (LOGGER.isLoggable(Level.WARNING)) {
                        LOGGER.log(
                            Level.WARNING,
                            "Failed to close the message service client.",
                            exception);
                    }
                }
            }
        }
    }

    public MessageServiceClient getMessageServiceClient() {
        return messageServiceClient;
    }

    public final void publish(final Serializable object) {
        currentMessagePublisher.publish(object);
    }

    public final void publish(final Serializable object, final Properties messageProperties) {
        currentMessagePublisher.publish(object, messageProperties);
    }

    public final void publish(final Serializable object, final Properties messageProperties, final String messageType) {
        currentMessagePublisher.publish(object, messageProperties, messageType);
    }

    public final void publish(final Serializable object, final String messageType) {
        currentMessagePublisher.publish(object, messageType);
    }

    public final void publish(final String text) {
        currentMessagePublisher.publish(text);
    }

    public final void publish(final String text, final Properties messageProperties) {
        currentMessagePublisher.publish(text, messageProperties);
    }

    public final void publish(final String text, final Properties messageProperties, final String messageType) {
        currentMessagePublisher.publish(text, messageProperties, messageType);
    }

    public final void publish(final String text, final String messageType) {
        currentMessagePublisher.publish(text, messageType);
    }

    public final void publishNow(final Serializable object) {
        currentMessagePublisher.publishNow(object);
    }

    public final void publishNow(final Serializable object, final Properties messageProperties) {
        currentMessagePublisher.publishNow(object, messageProperties);
    }

    public final void publishNow(final Serializable object, final Properties messageProperties, final String messageType) {
        currentMessagePublisher.publishNow(object, messageProperties, messageType);
    }

    public final void publishNow(final Serializable object, final String messageType) {
        currentMessagePublisher.publishNow(object, messageType);
    }

    public final void publishNow(final String text) {
        currentMessagePublisher.publishNow(text);
    }

    public final void publishNow(final String text, final Properties messageProperties) {
        currentMessagePublisher.publishNow(text, messageProperties);
    }

    public final void publishNow(final String text, final Properties messageProperties, final String messageType) {
        currentMessagePublisher.publishNow(text, messageProperties, messageType);
    }

    public final void publishNow(final String text, final String messageType) {
        currentMessagePublisher.publishNow(text, messageType);
    }

    public final void reconnect() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Reconnecting...");
        }
        if (scheduledThreadPoolExecutor == null) {
            // blocking reconnect
            reconnectNow();
        } else {
            // non-blocking reconnect
            new ReconnectTask().execute();
        }
    }

    public final boolean reconnectNow() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Reconnecting now...");
        }
        // blocking reconnect
        return new ReconnectTask().executeNow();
    }

    public final void setUp() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Setting up...");
        }
        setUp(
            !retryOnFail ? 0 : configuration.getAttributeAsInteger("interval", DEFAULT_INTERVAL),
            !retryOnFail ? 0 : configuration.getAttributeAsInteger("maxRetries", DEFAULT_MAX_RETRIES));
    }

    public final boolean setUpNow() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Setting up now...");
        }
        return
            setUpNow(
                !retryOnFail ? 0 : configuration.getAttributeAsInteger("interval", DEFAULT_INTERVAL),
                !retryOnFail ? 0 : configuration.getAttributeAsInteger("maxRetries", DEFAULT_MAX_RETRIES));
    }

    public final void start() {
        synchronized (stateLock) {
            requestedState = STATE_STARTED;
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(Level.FINE, "Requested State: STARTED");
            }
            if (currentState == STATE_SET_UP_DONE) {
                try {
                    // throws MessageServiceException
                    messageServiceClient.start();
                    currentState = STATE_STARTED;
                    if (LOGGER.isLoggable(Level.FINE)) {
                        LOGGER.log(Level.FINE, "Current State: STARTED");
                    }
                } catch (MessageServiceException exception) {
                    if (LOGGER.isLoggable(Level.SEVERE)) {
                        LOGGER.log(
                            Level.SEVERE,
                            "Failed to start message delivery!",
                            exception);
                    }
                }
            }
        }
    }

    public final void stop() {
        synchronized (stateLock) {
            requestedState = STATE_STOPPED;
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(Level.FINE, "Requested State: STOPPED");
            }
            if (currentState == STATE_STARTED) {
                try {
                    // throws MessageServiceException
                    messageServiceClient.stop();
                    currentState = STATE_STOPPED;
                    if (LOGGER.isLoggable(Level.FINE)) {
                        LOGGER.log(Level.FINE, "Current State: STOPPED");
                    }
                } catch (MessageServiceException exception) {
                    if (LOGGER.isLoggable(Level.SEVERE)) {
                        LOGGER.log(
                            Level.SEVERE,
                            "Failed to start message delivery!",
                            exception);
                    }
                }
            }
        }
    }

    public final void tearDown() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Tearing down...");
        }
        if (scheduledThreadPoolExecutor == null) {
            // blocking tear down
            tearDownNow();
        } else {
            // non-blocking tear down
            new TearDownTask(scheduledThreadPoolExecutor).execute();
        }
    }

    public final boolean tearDownNow() {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Tearing down now...");
        }
        // blocking tear down
        return new TearDownTask().executeNow();
    }

    protected void setUpMessageServiceClient()
    throws MessageServiceException {

    }

    protected void tearDownMessageServiceClient()                                                          
    throws MessageServiceException {

    }

    private void initialize()
    throws MessageServiceException {
        messageServiceClient.initialize();
    }

    private void setUp(final int interval, final int maxRetries) {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Setting up... (interval: [" + interval + "], maxRetries: [" + maxRetries + "])");
        }
        if (scheduledThreadPoolExecutor == null) {
            // blocking set up
            setUpNow(interval, maxRetries);
        } else {
            // non-blocking set up
            new SetUpTask(interval, maxRetries, scheduledThreadPoolExecutor).execute();
        }
    }

    private boolean setUpNow(final int interval, final int maxRetries) {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Setting up now... (interval: [" + interval + "], maxRetries: [" + maxRetries + "])");
        }
        // blocking set up
        return new SetUpTask(interval, maxRetries).executeNow();
    }

    private class ReconnectTask
    implements Runnable {
        private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

        private ScheduledFuture scheduledFuture;

        private boolean succeeded = false;

        private ReconnectTask() {
            this(null);
        }

        private ReconnectTask(final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
            this.scheduledThreadPoolExecutor = scheduledThreadPoolExecutor;
        }

        public void run() {
            try {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Reconnect task...");
                }
                long _failTimestamp = System.currentTimeMillis();
                synchronized (reconnectLock) {
                    if (_failTimestamp > successTimestamp + 5000) {
                        tearDownNow();
                        if (setUpNow(
                                !retryOnFail ?
                                    0 :
                                    configuration.getAttributeAsInteger(
                                        "intervalOnReconnect", DEFAULT_INTERVAL_ON_RECONNECT),
                                !retryOnFail ?
                                    0 :
                                    configuration.getAttributeAsInteger(
                                        "maxRetriesOnReconnect", DEFAULT_MAX_RETRIES_ON_RECONNECT))) {

                            successTimestamp = System.currentTimeMillis();
                            succeeded = true;
                        }
                    } else {
                        succeeded = true;
                    }
                }
                if (scheduledFuture != null) {
                    scheduledFuture.cancel(false);
                    scheduledFuture = null;
                }
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Reconnect task... (succeeded: [" + succeeded + "])");
                }
            } catch (final Exception exception) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(
                        Level.WARNING, "Exception caught on " + this.getClass().getName() + " Runnable.", exception);
                }
            }
        }

        private void execute() {
            if (scheduledThreadPoolExecutor != null) {
                scheduledFuture = scheduledThreadPoolExecutor.schedule(this, 0, TimeUnit.MILLISECONDS);
            } else {
                executeNow();
            }
        }

        private boolean executeNow() {
            run();
            return succeeded;
        }
    }

    private class SetUpTask
    implements Runnable {
        private final int interval;
        private final int maxRetries;
        private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

        private ScheduledFuture scheduledFuture;

        private boolean cancelled = false;
        private int retries = 0;
        private boolean succeeded = false;

        private SetUpTask(final int interval, final int maxRetries) {
            this(interval, maxRetries, null);
        }

        private SetUpTask(
            final int interval, final int maxRetries, final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {

            this.interval = interval;
            this.maxRetries = maxRetries;
            this.scheduledThreadPoolExecutor = scheduledThreadPoolExecutor;
        }

        public void run() {
            try {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Set Up task...");
                }
                try {
                    // throws MessageServiceException.
                    initialize();
                    // throws MessageServiceException
                    setUpMessageServiceClient();
                    cancel();
                    succeeded = true;
                    synchronized (stateLock) {
                        currentState = STATE_SET_UP_DONE;
                        if (LOGGER.isLoggable(Level.FINE)) {
                            LOGGER.log(Level.FINE, "Current State: SET UP DONE");
                        }
                        if (requestedState == STATE_STARTED) {
                            start();
                        }
                    }
                } catch (Exception exception) {
                    if (LOGGER.isLoggable(Level.FINE)) {
                        LOGGER.log(
                            Level.FINE, "Exception: " + exception.getClass().getName() + ": " + exception.getMessage());
                    }
                    tearDownNow();
                    if (retries++ == maxRetries) {
                        cancel();
                        if (LOGGER.isLoggable(Level.FINE)) {
                            LOGGER.log(
                                Level.FINE, "Set up of the message service client failed: " + exception.getMessage());
                        }
                        close();
                    }
                }
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Set Up task... (succeeded: [" + succeeded + "])");
                }
            } catch (final Exception exception) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(
                        Level.WARNING, "Exception caught on " + this.getClass().getName() + " Runnable.", exception);
                }
            }
        }

        private void cancel() {
            if (scheduledFuture != null) {
                scheduledFuture.cancel(false);
                scheduledFuture = null;
            }
            cancelled = true;
        }

        private void execute() {
            synchronized (stateLock) {
                currentState = STATE_SET_UP;
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Current State: SET UP");
                }
            }
            if (scheduledThreadPoolExecutor != null) {
                scheduledFuture =
                    scheduledThreadPoolExecutor.scheduleAtFixedRate(this, 0, interval, TimeUnit.MILLISECONDS);
            } else {
                executeNow();
            }
        }

        private boolean executeNow() {
            synchronized (stateLock) {
                currentState = STATE_SET_UP;
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Current State: SET UP");
                }
            }
            while (!cancelled) {
                run();
                if (!cancelled) {
                    synchronized (stateLock) {
                        if (requestedState == STATE_CLOSED) {
                            cancel();
                            break;
                        }
                    }
                    try {
                        // throws InterruptedException
                        Thread.sleep(interval);
                    } catch (InterruptedException exception) {
                        // do nothing
                    }
                    synchronized (stateLock) {
                        if (requestedState == STATE_CLOSED) {
                            cancel();
                            break;
                        }
                    }
                }
            }
            return succeeded;
        }
    }

    private class TearDownTask
    implements Runnable {
        private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

        private ScheduledFuture scheduledFuture;

        private boolean succeeded = false;

        private TearDownTask() {
            this(null);
        }

        private TearDownTask(final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
            this.scheduledThreadPoolExecutor = scheduledThreadPoolExecutor;
        }

        public void run() {
            try {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Tear Down task...");
                }
                try {
                    tearDownMessageServiceClient();
                    if (scheduledFuture != null) {
                        scheduledFuture.cancel(false);
                        scheduledFuture = null;
                    }
                    succeeded = true;
                    synchronized (stateLock) {
                        currentState = STATE_TEAR_DOWN_DONE;
                        if (LOGGER.isLoggable(Level.FINE)) {
                            LOGGER.log(Level.FINE, "Current State: TEAR DOWN DONE");
                        }
                        if (requestedState == STATE_CLOSED) {
                            close();
                        }
                    }
                } catch (Exception exception) {
                    if (scheduledFuture != null) {
                        scheduledFuture.cancel(false);
                        scheduledFuture = null;
                    }
                }
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Executing Tear Down task... (succeeded: [" + succeeded + "])");
                }
            } catch (final Exception exception) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(
                        Level.WARNING, "Exception caught on " + this.getClass().getName() + " Runnable.", exception);
                }
            }
        }

        private void execute() {
            synchronized (stateLock) {
                currentState = STATE_TEAR_DOWN;
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Current State: TEAR DOWN");
                }
            }
            if (scheduledThreadPoolExecutor != null) {
                scheduledFuture = scheduledThreadPoolExecutor.schedule(this, 0, TimeUnit.MILLISECONDS);
            } else {
                executeNow();
            }
        }

        private boolean executeNow() {
            synchronized (stateLock) {
                currentState = STATE_TEAR_DOWN;
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Current State: TEAR DOWN");
                }
            }
            run();
            return succeeded;
        }
    }
}
