/*
 * 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.http.Request;
import org.icepush.http.Response;
import org.icepush.http.ResponseHandler;
import org.icepush.http.Server;
import org.icepush.http.standard.FixedXMLContentHandler;
import org.icepush.http.standard.ResponseHandlerServer;
import org.icepush.util.Slot;

import java.io.*;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

public class BlockingConnectionServer extends TimerTask implements Server, NotificationBroadcaster.Receiver {
    private static final Logger log = Logger.getLogger(BlockingConnectionServer.class.getName());
    //Define here to avoid classloading problems after application exit
    private static final ResponseHandler ShutdownNoopResponse = new NoopResponseHandler("shutdown");
    private static final ResponseHandler TimeoutNoopResponse = new NoopResponseHandler("response timeout");
    private final ResponseHandler CloseResponseDup = new CloseConnectionResponseHandler("duplicate") {
        public void respond(Response response) throws Exception {
            super.respond(response);
            //revert timeout to previous value, duplicate requests can extend excessively the calculated delay
            //the duplicate requests occur during page reload or navigating away from the page and then returning back
            //before the server decides to sever the connection
            revertConnectionRecreationTimeout();
        }
    };
    private final ResponseHandler CloseResponseDown = new CloseConnectionResponseHandler("shutdown");
    private final Server AfterShutdown = new ResponseHandlerServer(CloseResponseDown);

    private final Lock backOffLock = new ReentrantLock();

    private final BlockingQueue<Request> pendingRequestQueue = new LinkedBlockingQueue<Request>(1);
    private final Slot heartbeatInterval;
    // This is either a LocalPushGroupManager or a DynamicPushGroupManager
    private final PushGroupManager pushGroupManager;
    private String browserID;
    private long responseTimeoutTime;
    private Server activeServer;
    private final Lock notifiedPushIDQueueLock = new ReentrantLock();
    private final Queue<String> notifiedPushIDQueue = new LinkedList<String>();
    private Set<String> participatingPushIDSet = Collections.emptySet();
    private Timer monitoringScheduler;

    private String lastWindow = "";
    private Set<String> lastNotifiedPushIDSet = new HashSet<String>();
    private NotifyBackURI notifyBackURI;
    private long defaultConnectionRecreationTimeout;
    private long responseTimestamp = System.currentTimeMillis();
    private long requestTimestamp = System.currentTimeMillis();
    private long backOffDelay = 0;

    public BlockingConnectionServer(final PushGroupManager pushGroupManager, final Timer monitoringScheduler, Slot heartbeat, final boolean terminateBlockingConnectionOnShutdown, Configuration configuration) {
        this.heartbeatInterval = heartbeat;
        this.pushGroupManager = pushGroupManager;
        this.defaultConnectionRecreationTimeout = configuration.getAttributeAsLong("connectionRecreationTimeout", 5000);
        this.monitoringScheduler = monitoringScheduler;
        //add monitor
        this.monitoringScheduler.scheduleAtFixedRate(this, 0, 1000);
        this.pushGroupManager.addBlockingConnectionServer(this);
        this.pushGroupManager.addNotificationReceiver(this);

        //define blocking server
        activeServer = new RunningServer(pushGroupManager, terminateBlockingConnectionOnShutdown);
    }

    public void backOff(final String browserID, final long delay) {
        getBackOffLock().lock();
        try {
            if (this.browserID != null && this.browserID.equals(browserID) && delay > 0) {
                backOffDelay = delay;
                respondToIfBackOffRequested((Request)null);
            }
        } finally {
            getBackOffLock().unlock();
        }
    }

    public void service(final Request request) throws Exception {
        activeServer.service(request);
    }

    public void shutdown() {
        cancel();
        pushGroupManager.deleteNotificationReceiver(this);
        pushGroupManager.removeBlockingConnectionServer(this);
        activeServer.shutdown();
    }

    public void run() {
        try {
            if ((System.currentTimeMillis() > responseTimeoutTime) && (!getPendingRequestQueue().isEmpty())) {
                // Respond only if there is a pending request as this is done on a non-container thread.
                respondIfPendingRequest(TimeoutNoopResponse);
            }
        } catch (Exception exception) {
            if (log.isLoggable(Level.WARNING)) {
                log.log(Level.WARNING, "Exception caught on " + this.getClass().getName() + " TimerTask.", exception);
            }
        }
    }

    private boolean sendNotificationsTo(final Request request, final Set<String> notificationSet) {
        //stop sending notifications if pushID are not used anymore by the browser
        Set<String> pushIDSet = new HashSet<String>(notificationSet);
        pushIDSet.retainAll(participatingPushIDSet);
        boolean anyNotifications = !pushIDSet.isEmpty();
        if (anyNotifications) {
            getNotifiedPushIDQueueLock().lock();
            try {
                for (final String _pushID : pushIDSet) {
                    if (!getModifiableNotifiedPushIDQueue().contains(_pushID)) {
                        getModifiableNotifiedPushIDQueue().add(_pushID);
                    }
                }
            } finally {
                getNotifiedPushIDQueueLock().unlock();
            }
            resetTimeout();
            return respondToIfNotificationsAvailable(request);
        }
        return anyNotifications;
    }

    private boolean resendLastNotifications(final Request request) {
        return sendNotificationsTo(request, getModifiableLastNotifiedPushIDSet());
    }

    private boolean respondToIfBackOffRequested(final Request request) {
        getBackOffLock().lock();
        try {
            boolean result = false;
            if (backOffDelay > 0) {
                BackOffResponseHandler backOffResponseHandler = new BackOffResponseHandler(backOffDelay);
                if (request != null) {
                    respondTo(request, backOffResponseHandler);
                    result = true;
                } else {
                    result = respondIfPendingRequest(backOffResponseHandler);
                }
                if (result) {
                    backOffDelay = 0;
                }
            }
            return result;
        } finally {
            getBackOffLock().unlock();
        }
    }

    private boolean respondToIfNotificationsAvailable(final Request request) {
        getNotifiedPushIDQueueLock().lock();
        try {
            boolean result = false;
            if (!getModifiableNotifiedPushIDQueue().isEmpty()) {
                //save notifications, maybe they will need to be resent when blocking connection switches to another window
                setLastNotifiedPushIDSet(new HashSet<String>(getModifiableNotifiedPushIDQueue()));
                NotificationHandler notificationHandler =
                    new NotificationHandler(getModifiableLastNotifiedPushIDSet()) {
                        public void writeTo(Writer writer) throws IOException {
                            super.writeTo(writer);

                            if (log.isLoggable(Level.FINE)) {
                                log.log(
                                    Level.FINE,
                                    "Push Notifications available for PushIDs '" + getModifiableNotifiedPushIDQueue() + "', " +
                                        "trying to respond.");
                            }
                            pushGroupManager.clearPendingNotifications(participatingPushIDSet);
                            getNotifiedPushIDQueueLock().lock();
                            try {
                                getModifiableNotifiedPushIDQueue().removeAll(getModifiableLastNotifiedPushIDSet());
                            } finally {
                                getNotifiedPushIDQueueLock().unlock();
                            }
                        }
                    };
                if (request != null) {
                    respondTo(request, notificationHandler);
                    result = true;
                } else {
                    result = respondIfPendingRequest(notificationHandler);
                }
            }
            return result;
        } finally {
            getNotifiedPushIDQueueLock().unlock();
        }
    }

    private void resetTimeout() {
        responseTimeoutTime = System.currentTimeMillis() + heartbeatInterval.getLongValue();
    }

    private void respondTo(final Request request, final ResponseHandler handler) {
        try {
            recordResponseTime();
            request.respondWith(handler);
        } catch (final IOException exception) {
            log.log(
                Level.FINE,
                "Possible communication issue encountered while responding: " + exception.getMessage(),
                exception
            );
        } catch (final Exception exception) {
            log.log(
                Level.SEVERE,
                "Failed to respond to pending request: " + exception.getMessage(),
                exception
            );
        }
    }

    private boolean respondIfPendingRequest(ResponseHandler handler) {
        Request previousRequest = getPendingRequestQueue().poll();
        if (previousRequest != null) {
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "Pending request for PushIDs '" + participatingPushIDSet + "', trying to respond.");
            }
            respondTo(previousRequest, handler);
            return true;
        }
        return false;
    }

    private static class ServerErrorResponseHandler extends FixedXMLContentHandler {
        private String message;

        private ServerErrorResponseHandler(Throwable exception) {
            this.message = exception.getMessage();
        }

        public void respond(Response response) throws Exception {
            response.setStatus(503);
            super.respond(response);
        }

        public void writeTo(Writer writer) throws IOException {
            writer.write("<server-error message=\"");
            writer.write(message);
            writer.write("\"/>");
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "Sending server-error - " + message);
            }
        }
    }

    private static class NoopResponseHandler extends FixedXMLContentHandler {
        private final String reason;

        private NoopResponseHandler(final String reason) {
            this.reason = reason;
        }

        @Override
        public void respond(final Response response)
        throws Exception {
            response.setHeader("X-Connection-reason", reason);
            super.respond(response);
        }

        public void writeTo(Writer writer) throws IOException {
            writer.write("<noop/>");
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "Sending NoOp.");
            }
        }
    }

    private class CloseConnectionResponseHandler implements ResponseHandler {
        private String reason = "undefined";

        public CloseConnectionResponseHandler(String reason) {
            this.reason = reason;
        }

        public void respond(Response response) throws Exception {
            //let the bridge know that this blocking connection should not be re-initialized
            response.setHeader("X-Connection", "close");
            response.setHeader("X-Connection-reason", reason);
            response.setHeader("Content-Length", 0);
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "Close current blocking connection.");
            }
        }
    }

    private class NotificationHandler
    extends FixedXMLContentHandler {
        private Set<String> notificationSet;

        private NotificationHandler(final Set<String> notificationSet) {
            this.notificationSet = new HashSet<String>(notificationSet);
        }

        public void writeTo(Writer writer) throws IOException {
            writer.write("<notified-pushids>");
            boolean first = true;
            for (final String pushID : notificationSet) {
                if (!first) {
                    writer.write(' ');
                } else {
                    first = false;
                }
                writer.write(pushID);
            }
            writer.write("</notified-pushids>");
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "Sending Notified PushIDs '" + notificationSet + "'.");
            }
        }
    }

    public void receive(final Set<String> notificationSet) {
        sendNotificationsTo((Request)null, notificationSet);
    }

    private class RunningServer implements Server {
        private final PushGroupManager pushGroupManager;
        private final boolean terminateBlockingConnectionOnShutdown;

        public RunningServer(PushGroupManager pushGroupManager, boolean terminateBlockingConnectionOnShutdown) {
            this.pushGroupManager = pushGroupManager;
            this.terminateBlockingConnectionOnShutdown = terminateBlockingConnectionOnShutdown;
        }

        public void service(final Request request) throws Exception {
            resetTimeout();
            try {
                participatingPushIDSet =
                    new HashSet<String>(Arrays.asList(request.getParameterAsStrings("ice.pushid")));
                adjustConnectionRecreationTimeout(request);
                // Respond only if there is a pending request as it would be considered a duplicate.
                respondIfPendingRequest(CloseResponseDup);
                if (browserID == null) {
                    browserID = getBrowserID(request);
                }
                long sequenceNumber;
                try {
                    sequenceNumber = request.getHeaderAsLong("ice.push.sequence");
                } catch (final RuntimeException exception) {
                    sequenceNumber = 0;
                }
                pushGroupManager.recordListen(participatingPushIDSet, sequenceNumber);
                //resend notifications if the window owning the blocking connection has changed
                String currentWindow = request.getHeader("ice.push.window");
                currentWindow = currentWindow == null ? "" : currentWindow;
                boolean resend = !lastWindow.equals(currentWindow);
                lastWindow = currentWindow;
                setNotifyBackURI(request);
                pushGroupManager.cancelConfirmationTimeout(participatingPushIDSet);
                pushGroupManager.cancelExpiryTimeout(participatingPushIDSet);
                pushGroupManager.startExpiryTimeout(participatingPushIDSet, notifyBackURI);
                if (null != notifyBackURI)  {
                    pushGroupManager.pruneParkedIDs(notifyBackURI, participatingPushIDSet);
                }
                if (!respondToIfBackOffRequested(request)) {
                    // No response has been sent to the request.
                    if (!sendNotificationsTo(request, pushGroupManager.getPendingNotificationSet())) {
                        // No response has been sent to the request.
                        if (!resend || !resendLastNotifications(request)) {
                            if (!respondToIfNotificationsAvailable(request)) {
                                getPendingRequestQueue().put(request);
                            }
                        }
                    }
                }
            } catch (final Throwable throwable) {
                log.log(Level.WARNING, "Failed to respond to request", throwable);
                respondTo(request, new ServerErrorResponseHandler(throwable));
            }
        }

        public void shutdown() {
            //avoid creating new blocking connections after shutdown
            activeServer = AfterShutdown;
            respondIfPendingRequest(terminateBlockingConnectionOnShutdown ? CloseResponseDown : ShutdownNoopResponse);
        }
    }

    private void adjustConnectionRecreationTimeout(Request request) {
        for (final String pushIDString : participatingPushIDSet) {
            PushID pushID = pushGroupManager.getPushID(pushIDString);
            if (pushID != null) {
                if (pushID.getStatus().getConnectionRecreationTimeout() == -1) {
                    pushID.getStatus().setConnectionRecreationTimeout(defaultConnectionRecreationTimeout);
                }
                pushID.getStatus().backUpConnectionRecreationTimeout();
            }
        }
        long now = System.currentTimeMillis();
        long elapsed = now - requestTimestamp;
        requestTimestamp = now;
        long currentResponseDelay = requestTimestamp - responseTimestamp;
        //adaptive timeout -- see algorithm described in PUSH-164
        long responseDelay = currentResponseDelay;
        for (final String pushIDString : participatingPushIDSet) {
            PushID pushID = pushGroupManager.getPushID(pushIDString);
            if (pushID != null) {
                responseDelay = Math.max(responseDelay, (pushID.getStatus().getConnectionRecreationTimeout() * 4) / 5);
                responseDelay = Math.min(responseDelay, (pushID.getStatus().getConnectionRecreationTimeout() * 3) / 2);
                responseDelay = Math.max(responseDelay, 500);
                pushID.getStatus().setConnectionRecreationTimeout(
                    (responseDelay + (pushID.getStatus().getConnectionRecreationTimeout() * 4)) / 5);
            }
        }

        if (log.isLoggable(Level.FINE)) {
            String browserID = getBrowserID(request);
            if (null == browserID)  {
                browserID = "undefined";
            }
            participatingPushIDSet = new HashSet<String>(Arrays.asList(request.getParameterAsStrings("ice.pushid")));
            setNotifyBackURI(request);
            log.log(
                Level.FINE,
                "ICEpush metric:" +
                    " IP: " + request.getRemoteAddr() +
                    " pushIds: " + participatingPushIDSet +
                    " Cloud Push ID: " + notifyBackURI +
                    " Browser: " + browserID +
                    " last request: " + elapsed +
                    " Latency: " + currentResponseDelay);
        }
    }

    public static String getBrowserID(Request request) {
        return request.getParameter("ice.push.browser");
    }

    protected Lock getBackOffLock() {
        return backOffLock;
    }

    protected Set<String> getModifiableLastNotifiedPushIDSet() {
        return lastNotifiedPushIDSet;
    }

    protected Queue<String> getModifiableNotifiedPushIDQueue() {
        return notifiedPushIDQueue;
    }

    protected Lock getNotifiedPushIDQueueLock() {
        return notifiedPushIDQueueLock;
    }

    protected BlockingQueue<Request> getPendingRequestQueue() {
        return pendingRequestQueue;
    }

    protected void setLastNotifiedPushIDSet(final Set<String> lastNotifiedPushIDSet) {
        this.lastNotifiedPushIDSet = lastNotifiedPushIDSet;
    }

    private void revertConnectionRecreationTimeout() {
        for (final String pushIDString : participatingPushIDSet) {
            PushID pushID = pushGroupManager.getPushID(pushIDString);
            if (pushID != null) {
                pushID.getStatus().revertConnectionRecreationTimeout();
            }
        }
    }

    private void recordResponseTime() {
        responseTimestamp = System.currentTimeMillis();
    }

    private void setNotifyBackURI(final Request request) {
        String notifyBack = request.getHeader("ice.notifyBack");
        if (notifyBack != null && notifyBack.trim().length() != 0) {
            if (this.notifyBackURI == null || !this.notifyBackURI.getURI().equals(notifyBack)) {
                log.log(Level.FINE, "Found new NotifyBackURI on Request: '" + notifyBack + "'.");
                this.notifyBackURI = new NotifyBackURI(notifyBack);
            }
            pushGroupManager.setNotifyBackURI(participatingPushIDSet, this.notifyBackURI, true);
        } else {
            //If notifyBackURI was set on the server side with no corresponding request from the client
            //then we should query the PushGroupManager for the uri and set it here.
            Iterator<String> idIterator = participatingPushIDSet.iterator();
            while (idIterator.hasNext()) {
                NotifyBackURI notifyBackURI = pushGroupManager.getNotifyBackURI(idIterator.next());
                if (notifyBackURI != null) {
                    log.log(Level.FINE, "Found NotifyBackURI on PushGroupManager: '" + notifyBackURI + "'.");
                    this.notifyBackURI = notifyBackURI;
                    break;
                }
            }
        }
    }
}
