/*
 * Copyright 2004-2014 ICEsoft Technologies Canada Corp.
 *
 * 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 org.icepush;

import org.icepush.servlet.ServletContextConfiguration;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.servlet.ServletContext;

public class LocalPushGroupManager extends AbstractPushGroupManager implements PushGroupManager {
    private static final Logger LOGGER = Logger.getLogger(LocalPushGroupManager.class.getName());
    public static final int DefaultNotificationQueueSize = 1000;
    public static final int DefaultMinCloudPushInterval = 10 * 1000;
    public static final int DefaultCloudPushIdTimeout = 30 * 60 * 1000;
    public static final int DefaultPushIdTimeout = 2 * 60 * 1000;     
    private static final String[] STRINGS = new String[0];
    private static final OutOfBandNotifier NOOPOutOfBandNotifier = new OutOfBandNotifier() {
        public void broadcast(PushNotification notification, String[] uris) {
            System.out.println("message send " + notification + " to " + Arrays.asList(uris));
        }

        public void registerProvider(String protocol, NotificationProvider provider) {
        }
        public void trace(String message)  {
            if (LOGGER.isLoggable(Level.WARNING)) {
                LOGGER.log(Level.WARNING, "NOOPOutOfBandNotifier discarding notification " + message);
            }
        }
    };
    private final Notification NOOP = new Notification("---") {
        public void run() {
        }
    };
    private final Set<BlockingConnectionServer> blockingConnectionServerSet =
        new CopyOnWriteArraySet<BlockingConnectionServer>();
    protected final ConcurrentMap<String, PushID> pushIDMap = new ConcurrentHashMap<String, PushID>();
    protected final ConcurrentMap<String, Group> groupMap = new ConcurrentHashMap<String, Group>();
    private final ConcurrentMap<String, NotifyBackURI> parkedPushIDs = new ConcurrentHashMap<String, NotifyBackURI>();
    private final Lock pendingNotifiedPushIDSetLock = new ReentrantLock();
    private final Set<String> pendingNotifiedPushIDSet = new HashSet<String>();
    private final NotificationBroadcaster outboundNotifier = new LocalNotificationBroadcaster();
    private final Timer timer = new Timer("Notification queue consumer.", true);
    private final TimerTask queueConsumer;
    private final Lock notificationQueueLock = new ReentrantLock();
    private final Condition notificationAvailableCondition = getNotificationQueueLock().newCondition();
    private final Queue<Notification> notificationQueue;
    private final long pushIdTimeout;
    private final long cloudPushIdTimeout;
    private final long minCloudPushInterval;
    private final ServletContext context;
    private final Timer timeoutTimer = new Timer("Timeout timer", true);

    public LocalPushGroupManager(final ServletContext servletContext) {
        this.context = servletContext;
        Configuration configuration = new ServletContextConfiguration("org.icepush", servletContext);
        this.pushIdTimeout = configuration.getAttributeAsLong("pushIdTimeout", DefaultPushIdTimeout);
        this.cloudPushIdTimeout = configuration.getAttributeAsLong("cloudPushIdTimeout", DefaultCloudPushIdTimeout);
        this.minCloudPushInterval = configuration.getAttributeAsLong("minCloudPushInterval", DefaultMinCloudPushInterval);
        int notificationQueueSize = configuration.getAttributeAsInteger("notificationQueueSize", DefaultNotificationQueueSize);
        this.notificationQueue = new LinkedBlockingQueue<Notification>(notificationQueueSize);        
        this.queueConsumer = new QueueConsumerTask();
        this.timer.schedule(queueConsumer, 0);
    }


    public void addBlockingConnectionServer(final BlockingConnectionServer server) {
        blockingConnectionServerSet.add(server);
    }

    public void addMember(final String groupName, final String id) {
        if (groupName != null && id != null) {
            PushID pushID = pushIDMap.get(id);
            if (pushID == null) {
                pushIDMap.put(id, newPushID(id, groupName));
            } else {
                pushID.addToGroup(groupName);
            }
            Group group = groupMap.get(groupName);
            if (group == null) {
                groupMap.put(groupName, new Group(groupName, id, this));
            } else {
                group.addPushID(id);
            }
            memberAdded(groupName, id);
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(Level.FINE, "Added PushID '" + id + "' to Push Group '" + groupName + "'.");
            }
        }
    }

    public void addNotificationReceiver(final NotificationBroadcaster.Receiver observer) {
        outboundNotifier.addReceiver(observer);
    }

    public void backOff(final String browserID, final long delay) {
        for (final BlockingConnectionServer server : blockingConnectionServerSet) {
            server.backOff(browserID, delay);
        }
    }

    public void deleteNotificationReceiver(final NotificationBroadcaster.Receiver observer) {
        outboundNotifier.deleteReceiver(observer);
    }

    public void clearPendingNotification(final String pushID) {
        getPendingNotifiedPushIDSetLock().lock();
        try {
            getModifiablePendingNotifiedPushIDSet().remove(pushID);
        } finally {
            getPendingNotifiedPushIDSetLock().unlock();
        }
    }

    public void clearPendingNotifications(final Set<String> pushIDSet) {
        getPendingNotifiedPushIDSetLock().lock();
        try {
            getModifiablePendingNotifiedPushIDSet().removeAll(pushIDSet);
        } finally {
            getPendingNotifiedPushIDSetLock().unlock();
        }
    }

    public Set<String> getPendingNotificationSet() {
        getPendingNotifiedPushIDSetLock().lock();
        try {
            return new HashSet<String>(getPendingNotifiedPushIDSet());
        } finally {
            getPendingNotifiedPushIDSetLock().unlock();
        }
    }

    public Set<String> getPendingNotifiedPushIDSet() {
        return Collections.unmodifiableSet(getModifiablePendingNotifiedPushIDSet());
    }

    public Map<String, String[]> getGroupMap() {
        Map<String, String[]> groupMap = new HashMap<String, String[]>();
        for (Group group : new ArrayList<Group>(this.groupMap.values())) {
            groupMap.put(group.getName(), group.getPushIDs());
        }
        return groupMap;
    }

    public void push(final String groupName) {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Push Notification request for Push Group '" + groupName + "'.");
        }
        Notification _notification = new Notification(groupName);
        getNotificationQueueLock().lock();
        try {
            if (!getNotificationQueue().contains(_notification)) {
                if (getNotificationQueue().offer(_notification)) {
                    getNotificationAvailableCondition().signalAll();
                } else {
                    // Leave at INFO
                    if (LOGGER.isLoggable(Level.INFO)) {
                        LOGGER.log(
                            Level.INFO,
                            "Push Notification request for Push Group '" + groupName + "' was dropped, " +
                                "queue maximum size reached.");
                    }
                }
            } else {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(
                        Level.FINE,
                        "Push Notification request for Push Group '" + groupName + "' was ignored, duplication.");
                }
            }
        } finally {
            getNotificationQueueLock().unlock();
        }
    }

    public void push(final String groupName, final PushConfiguration pushConfiguration) {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(
                Level.FINE,
                "Push Notification request for Push Group '" + groupName + "' " +
                    "(Push configuration: '" + pushConfiguration + "').");
        }
        Notification _notification;
        if (pushConfiguration.getAttributes().get("subject") != null) {
            _notification = new OutOfBandNotification(groupName, pushConfiguration);
        } else {
            _notification = new Notification(groupName, pushConfiguration);
        }
        getNotificationQueueLock().lock();
        try {
            if (!getNotificationQueue().contains(_notification)) {
                //add this notification to a blocking queue
                getNotificationQueue().add(_notification);
                getNotificationAvailableCondition().signalAll();
            } else {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(
                        Level.FINE,
                        "Push Notification request for Push Group '" + groupName + "' was ignored, duplication.");
                }
            }
        } finally {
            getNotificationQueueLock().unlock();
        }
    }

    public void recordListen(final Set<String> pushIDSet, final long sequenceNumber) {
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(
                        Level.FINE,
                        "Record listen for PushID '" + pushIDString + "' (Sequence# " + sequenceNumber + ").");
                }
                pushID.setSequenceNumber(sequenceNumber);
            }
        }
    }

    public void removeBlockingConnectionServer(final BlockingConnectionServer server) {
        blockingConnectionServerSet.remove(server);
    }

    public void removeMember(final String groupName, final String pushID) {
        if (groupName != null && pushID != null) {
            Group group = groupMap.get(groupName);
            if (group != null) {
                group.removePushID(pushID);
                PushID id = pushIDMap.get(pushID);
                if (id != null) {
                    id.removeFromGroup(groupName);
                }
                memberRemoved(groupName, pushID);
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Removed PushID '" + pushID + "' from Push Group '" + groupName + "'.");
                }
            }
        }
    }

    public boolean setNotifyBackURI(
        final Set<String> pushIDSet, final NotifyBackURI notifyBackURI, final boolean broadcast) {

        boolean modified = false;
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                if (pushID.setNotifyBackURI(notifyBackURI)) {
                    modified = true;
                } else {
                    pushID.getNotifyBackURI().touch();
                }
            }
        }
        return modified;
    }

    public void park(final String pushId, final NotifyBackURI notifyBackURI) {
        parkedPushIDs.put(pushId, notifyBackURI);
    }

    public void pruneParkedIDs(final NotifyBackURI notifyBackURI, final Set<String> listenedPushIDSet) {
        for (final Map.Entry<String, NotifyBackURI> parkedPushIDEntry : parkedPushIDs.entrySet()) {
            String parkedPushID = parkedPushIDEntry.getKey();
            if (parkedPushIDEntry.getValue().getURI().equals(notifyBackURI.getURI()) &&
                !listenedPushIDSet.contains(parkedPushID)) {

                parkedPushIDs.remove(parkedPushID);
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(
                        Level.FINE,
                        "Removed unlistened parked PushID '" + parkedPushID + "' for " +
                            "NotifyBackURI '" + notifyBackURI + "'.");
                }
            }
        }
    }

    public void shutdown() {
        queueConsumer.cancel();
        timer.cancel();
        timeoutTimer.cancel();
    }

    public void cancelConfirmationTimeout(final Set<String> pushIDSet) {
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                pushID.cancelConfirmationTimeout();
            }
        }
    }

    public void cancelExpiryTimeout(final Set<String> pushIDSet) {
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                pushID.cancelExpiryTimeout();
            }
        }
    }

    public void startConfirmationTimeout(final Set<String> pushIDSet) {
        Set<NotifyBackURI> notifyBackURISet = new HashSet<NotifyBackURI>();
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                NotifyBackURI notifyBackURI = pushID.getNotifyBackURI();
                if (notifyBackURI != null && !notifyBackURISet.contains(notifyBackURI)) {
                    notifyBackURISet.add(notifyBackURI);
                    pushID.startConfirmationTimeout(pushID.getSequenceNumber());
                }
            }
        }
    }

    public void startExpiryTimeout(final Set<String> pushIDSet, final NotifyBackURI notifyBackURI) {
        for (final String pushIDString : pushIDSet) {
            PushID pushID = pushIDMap.get(pushIDString);
            if (pushID != null) {
                pushID.startExpiryTimeout(notifyBackURI, pushID.getSequenceNumber());
            }
        }
    }

    public Map<String, PushID> getPushIDMap() {
        return Collections.unmodifiableMap(pushIDMap);
    }

    public PushID getPushID(final String pushID) {
        return pushIDMap.get(pushID);
    }

    protected long getCloudPushIDTimeout() {
        return cloudPushIdTimeout;
    }

    protected long getMinCloudPushInterval() {
        return minCloudPushInterval;
    }

    protected Set<String> getModifiablePendingNotifiedPushIDSet() {
        return pendingNotifiedPushIDSet;
    }

    protected Condition getNotificationAvailableCondition() {
        return notificationAvailableCondition;
    }

    protected Queue<Notification> getNotificationQueue() {
        return notificationQueue;
    }

    protected Lock getNotificationQueueLock() {
        return notificationQueueLock;
    }

    protected Lock getPendingNotifiedPushIDSetLock() {
        return pendingNotifiedPushIDSetLock;
    }

    protected long getPushIDTimeout() {
        return pushIdTimeout;
    }

    protected Timer getTimeoutTimer() {
        return timeoutTimer;
    }

    protected PushID newPushID(final String id, final String groupName) {
        PushID pushID =
            new PushID(
                id, groupName, getPushIDTimeout(), getCloudPushIDTimeout(), getTimeoutTimer(),
                getMinCloudPushInterval(), this);
        pushID.startExpiryTimeout();
        return pushID;
    }

    OutOfBandNotifier getOutOfBandNotifier() {
        Object attribute = context.getAttribute(OutOfBandNotifier.class.getName());
        return attribute == null ? NOOPOutOfBandNotifier : (OutOfBandNotifier) attribute;
    }

    Group getGroup(final String groupName) {
        return groupMap.get(groupName);
    }

    public NotifyBackURI getNotifyBackURI(final String pushID) {
        NotifyBackURI parkedURI = parkedPushIDs.get(pushID);
        if(parkedURI != null){
            return parkedURI;
        }
        //The pushid that has a registered notifyBackURI is not always parked.
        PushID pusher = pushIDMap.get(pushID);
        NotifyBackURI pushURI = null;
        if( pusher != null){
            pushURI = pusher.getNotifyBackURI();
        }
        return pushURI;
    }

    boolean isParked(final String pushID) {
        return parkedPushIDs.containsKey(pushID);
    }

    void removeGroup(final String groupName) {
        groupMap.remove(groupName);
    }

    void removePendingNotification(final String pushID) {
        clearPendingNotification(pushID);
    }

    void removePendingNotifications(final Set<String> pushIDSet) {
        clearPendingNotifications(pushIDSet);
    }

    void removePushID(final String pushID) {
        pushIDMap.remove(pushID);
    }

    private class Notification
    implements Runnable {
        protected final Set<String> exemptPushIDSet = new HashSet<String>();

        protected final String groupName;

        public Notification(final String groupName) {
            this.groupName = groupName;
        }

        public Notification(final String groupName, final PushConfiguration pushConfiguration) {
            this.groupName = groupName;
            Set<String> pushIDSet = (Set<String>)pushConfiguration.getAttributes().get("pushIDSet");
            if (pushIDSet != null) {
                getExemptPushIDSet().addAll(pushIDSet);
            }
        }

        @Override
        public boolean equals(final Object object) {
            return
                object instanceof Notification &&
                    ((Notification)object).getExemptPushIDSet().containsAll(getExemptPushIDSet()) &&
                    ((Notification)object).getExemptPushIDSet().size() == getExemptPushIDSet().size() &&
                    ((Notification)object).getGroupName().equals(getGroupName());
        }

        @Override
        public int hashCode() {
            return getGroupName().hashCode();
        }

        public void run() {
            Group group = groupMap.get(getGroupName());
            if (group != null) {
                if (LOGGER.isLoggable(Level.FINE)) {
                    LOGGER.log(Level.FINE, "Push Notification triggered for Push Group '" + getGroupName() + "'.");
                }
                Set<String> pushIDSet = new HashSet<String>(Arrays.asList(group.getPushIDs()));
                pushIDSet.removeAll(exemptPushIDSet);
                getPendingNotifiedPushIDSetLock().lock();
                try {
                    getModifiablePendingNotifiedPushIDSet().addAll(pushIDSet);
                } finally {
                    getPendingNotifiedPushIDSetLock().unlock();
                }
                startConfirmationTimeout(pushIDSet);
                outboundNotifier.broadcast(pushIDSet);
                pushed(getGroupName());
            }
        }

        @Override
        public String toString() {
            return
                new StringBuilder().
                    append("Notification[").
                        append(membersAsString()).
                    append("]").
                        toString();
        }

        protected Set<String> getExemptPushIDSet() {
            return exemptPushIDSet;
        }

        protected String getGroupName() {
            return groupName;
        }

        protected String membersAsString() {
            return
                new StringBuilder().
                    append("exemptPushIDSet: '").append(getExemptPushIDSet()).append("', ").
                    append("groupName: '").append(getGroupName()).append("'").
                        toString();
        }
    }

    private class OutOfBandNotification
    extends Notification
    implements Runnable {
        private final PushConfiguration pushConfiguration;

        public OutOfBandNotification(final String groupName, final PushConfiguration pushConfiguration) {
            super(groupName, pushConfiguration);
            this.pushConfiguration = pushConfiguration;
        }

        @Override
        public boolean equals(final Object object) {
            return
                object instanceof OutOfBandNotification &&
                    ((OutOfBandNotification)object).getPushConfiguration().equals(getPushConfiguration()) &&
                    super.equals(object);
        }

        public void run() {
            Group group = groupMap.get(getGroupName());
            String[] pushIDs = STRINGS;
            if (group != null) {
                pushIDs = group.getPushIDs();
            }
            for (final String pushIDString : pushIDs) {
                PushID pushID = pushIDMap.get(pushIDString);
                if (pushID != null) {
                    pushID.setPushConfiguration(getPushConfiguration());
                }
            }
            super.run();
        }

        @Override
        public String toString() {
            return
                new StringBuilder().
                    append("OutOfBandNotification[").
                        append(membersAsString()).
                    append("]").
                        toString();
        }

        protected PushConfiguration getPushConfiguration() {
            return pushConfiguration;
        }

        @Override
        protected String membersAsString() {
            return
                new StringBuilder().
                    append("pushConfiguration: '").append(getPushConfiguration()).append("', ").
                    append(super.membersAsString()).
                        toString();
        }
    }

    private class QueueConsumerTask extends TimerTask {
        private boolean running = true;

        public void run() {
            try {
                //take tasks from the queue and execute them serially
                while (running) {
                    getNotificationQueueLock().lock();
                    try {
                        if (getNotificationQueue().isEmpty()) {
                            try {
                                getNotificationAvailableCondition().await();
                            } catch (final InterruptedException exception) {
                                LOGGER.log(Level.FINE, "Notification queue draining interrupted.");
                            }
                        }
                        getNotificationQueue().poll().run();
                    } catch (final Throwable throwable)  {
                        LOGGER.log(Level.WARNING, "Notification queue encountered ", throwable);
                    } finally {
                        getNotificationQueueLock().unlock();
                    }
                }
            } catch (final Exception exception) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(
                        Level.WARNING, "Exception caught on " + this.getClass().getName() + " TimerTask.", exception);
                }
            }
        }

        public boolean cancel() {
            running = false;
            getNotificationQueueLock().lock();
            try {
                getNotificationQueue().offer(NOOP);//add noop to unblock the queue
                getNotificationAvailableCondition().signalAll();
            } finally {
                getNotificationQueueLock().unlock();
            }
            return super.cancel();
        }
    }
}
