/*
 *  Copyright 1999-2004 The Apache Software Foundation 
 *
 * 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.apache.slide.cluster;

import java.util.Enumeration;
import java.util.EventListener;
import java.util.Iterator;
import java.util.Map;

import org.apache.commons.httpclient.Credentials;
import org.apache.commons.httpclient.UsernamePasswordCredentials;
import org.apache.commons.httpclient.protocol.Protocol;
import org.apache.slide.authenticate.CredentialsToken;
import org.apache.slide.authenticate.SecurityToken;
import org.apache.slide.common.Domain;
import org.apache.slide.common.NamespaceAccessToken;
import org.apache.slide.common.SlideTokenImpl;
import org.apache.slide.common.Uri;
import org.apache.slide.store.ExtendedStore;
import org.apache.slide.store.Store;
import org.apache.slide.util.conf.Configurable;
import org.apache.slide.util.conf.Configuration;
import org.apache.slide.util.conf.ConfigurationException;
import org.apache.slide.util.logger.Logger;
import org.apache.webdav.lib.NotificationListener;
import org.apache.webdav.lib.Subscriber;
import org.apache.webdav.lib.methods.DepthSupport;

/**
 * <h3>Description</h3>
 * <p>
 * When configured properly this class will register with one or more external
 * Slide instances and listen for changes. Upon notification of a change this
 * class will cause the cache of the local Slide instance to be refreshed for
 * the changed object.
 * </p>
 * <h3>Usage</h3>
 * <p>
 * Add the following to your Domain.xml inside the &lt;events&gt; node.
 * </p>
 * 
 * <pre>
 * 
 *  
 *  	&lt;listener classname=&quot;org.apache.slide.cluster.ClusterCacheRefresher&quot;&gt;
 *  		&lt;configuration&gt;
 *  			&lt;node local-host=&quot;local.host.domain&quot;
 *  			      local-port=&quot;4444&quot;
 *  				  repository-host=&quot;remote.host.domain&quot;
 *  				  repository-port=&quot;8080&quot;
 *  				  repository-protocol=&quot;http&quot;
 *  				  username=&quot;root&quot;
 *  				  password=&quot;root&quot;
 *  			/&gt;
 *  		&lt;/configuration&gt;
 *  	&lt;/listener&gt;
 *    
 *  
 * </pre>
 * 
 * <p>
 * There should be one &lt;node&gt; element for each node in the cluster,
 * <b>except </b> for the current node. ClusterCacheRefresher should not be
 * configured to listen to itself except for testing purposes.
 * </p>
 * <h3>&lt;node&gt; attributes</h3>
 * <table>
 * <tr>
 * <th>Attribute Name</th>
 * <th>Required?</th>
 * <th>Default Value</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>local-host</td>
 * <td>yes</td>
 * <td>none</td>
 * <td>A network-accessible name or ip-address where the remote Slide instance
 * can reach <b>this server. </b></td>
 * </tr>
 * <tr>
 * <td>local-port</td>
 * <td>yes</td>
 * <td>none</td>
 * <td>A port number ClusterCacheRefresher can use to listen for notifications.
 * <b>Must be unique. </b></td>
 * </tr>
 * <tr>
 * <td>repository-host</td>
 * <td>yes</td>
 * <td>none</td>
 * <td>A network-accessible name or ip-address of the remote Slide instance to
 * monitor.</td>
 * </tr>
 * <tr>
 * <td>repository-port</td>
 * <td>yes</td>
 * <td>none</td>
 * <td>The port the remote Slide instance is running on.</td>
 * </tr>
 * <tr>
 * <td>repository-protocol</td>
 * <td>no</td>
 * <td>http</td>
 * <td>The protocol the remote Slide instance is using. Must be one of "http"
 * or "https".</td>
 * </tr>
 * <tr>
 * <td>username</td>
 * <td>no</td>
 * <td>none</td>
 * <td>The username to use to connect to the remote Slide instance.</td>
 * </tr>
 * <tr>
 * <td>password</td>
 * <td>no</td>
 * <td>none</td>
 * <td>The password that goes with the username.</td>
 * </tr>
 * <tr>
 * <td>repository-domain</td>
 * <td>no</td>
 * <td>/slide</td>
 * <td>The context path of the remote Slide instance.</td>
 * </tr>
 * <tr>
 * <td>poll-interval</td>
 * <td>no</td>
 * <td>60000</td>
 * <td>The number of milliseconds to wait between polling the remote Slide
 * instance for any changes. Polling for changes is a backup only, so this value
 * can be set fairly high.</td>
 * </tr>
 * <tr>
 * <td>udp</td>
 * <td>no</td>
 * <td>true</td>
 * <td>Must be "true" or "false". Indicates whether to use udp or tcp to listen
 * for notifications.</td>
 * </tr>
 * <tr>
 * <td>base-uri</td>
 * <td>no</td>
 * <td>/</td>
 * <td>The base path to monitor for changes. Will be appended to the
 * repository-domain.</td>
 * </tr>
 * <tr>
 * <td>subscription-lifetime</td>
 * <td>no</td>
 * <td>3600</td>
 * <td>The number of seconds a subscription should last. Subscriptions are
 * automatically refreshed. Do not set this value too high.</td>
 * </tr>
 * <tr>
 * <td>notification-delay</td>
 * <td>no</td>
 * <td>0</td>
 * <td>Number of seconds the remote Slide instance should wait before sending a
 * notification of a change.</td>
 * </tr>
 * </table>
 */
public class ClusterCacheRefresher implements EventListener, Configurable {
    protected static final String LOG_CHANNEL = ClusterCacheRefresher.class.getName();

    protected NotificationListener listener;

    public ClusterCacheRefresher() {
        Domain.log("Creating ClusterCacheRefresher", LOG_CHANNEL, Logger.INFO);
    }

    public void configure(Configuration configuration) throws ConfigurationException {
        Domain.log("Configuring ClusterCacheRefresher", LOG_CHANNEL, Logger.INFO);

        Enumeration nodes = configuration.getConfigurations("node");
        while (nodes.hasMoreElements()) {
            Configuration node = (Configuration) nodes.nextElement();
            final String host = node.getAttribute("local-host");
            final int port = node.getAttributeAsInt("local-port");
            final String repositoryHost = node.getAttribute("repository-host");
            final int repositoryPort = node.getAttributeAsInt("repository-port");
            String repositoryProtocolString = node.getAttribute("repository-protocol", "http");
            final Protocol protocol;
            try {
                protocol = Protocol.getProtocol(repositoryProtocolString);
            } catch (IllegalStateException exception) {
                throw new ConfigurationException("Unknown repository-protocol: " + repositoryProtocolString
                        + ". Must be \"http\" or \"https\".", configuration);
            }
            String username = node.getAttribute("username", "");
            String password = node.getAttribute("password", "");
            final Credentials credentials = new UsernamePasswordCredentials(username, password);
            final String repositoryDomain = node.getAttribute("repository-domain", "/slide");
            final int pollInterval = node.getAttributeAsInt("poll-interval", 60000);
            final boolean udp = node.getAttributeAsBoolean("udp", true);
            final String uri = node.getAttribute("base-uri", "/");
            final int depth = DepthSupport.DEPTH_INFINITY;
            final int lifetime = node.getAttributeAsInt("subscription-lifetime", 3600);
            final int notificationDelay = node.getAttributeAsInt("notification-delay", 0);

            final Subscriber contentSubscriber = new Subscriber() {
                public void notify(String uri, Map information) {
                    NamespaceAccessToken nat = Domain.accessNamespace(new SecurityToken(this), Domain.getDefaultNamespace());
                    try {
                        nat.begin();
                        Iterator keys = information.keySet().iterator();
                        while (keys.hasNext()) {
                            String key = keys.next().toString();
                            if ("uri".equals(key)) {
                                Uri theUri = nat.getUri(new SlideTokenImpl(new CredentialsToken("")), stripUri(information.get(key).toString()));
                                Store store = theUri.getStore();
                                if (store instanceof ExtendedStore) {
                                    Domain.log("Resetting cache for " + theUri, LOG_CHANNEL, Logger.INFO);
                                    ((ExtendedStore) store).removeObjectFromCache(theUri);
                                }
                            }
                        }
                        nat.commit();
                    } catch(Exception e) {
                    	if (Domain.isEnabled(LOG_CHANNEL, Logger.ERROR)) {
                    		Domain.log("Error clearing cache: " + e + ". See stderr for stacktrace.", LOG_CHANNEL, Logger.ERROR);
                    		e.printStackTrace();
                    	}
                    }
                }
            };

            final Subscriber structureSubscriber = new Subscriber() {
                public void notify(String uri, Map information) {
                    NamespaceAccessToken nat = Domain.accessNamespace(new SecurityToken(this), Domain.getDefaultNamespace());
                    try {
                        nat.begin();
	                    Iterator keys = information.keySet().iterator();
	                    while (keys.hasNext()) {
	                        String key = keys.next().toString();
	                        if ("uri".equals(key)) {
	                            Uri theUri = nat.getUri(new SlideTokenImpl(new CredentialsToken("")), stripUri(information.get(key).toString()));
	                            Store store = theUri.getParentUri().getStore();
	                            if (store instanceof ExtendedStore) {
	                                Domain.log("Resetting cache for " + theUri.getParentUri(), LOG_CHANNEL, Logger.INFO);
	                                ((ExtendedStore) store).removeObjectFromCache(theUri.getParentUri());
	                            }
	                        }
	                    }
                        nat.commit();
                    } catch(Exception e) {
                    	if (Domain.isEnabled(LOG_CHANNEL, Logger.ERROR)) {
                    		Domain.log("Error clearing cache: " + e + ". See stderr for stacktrace.", LOG_CHANNEL, Logger.ERROR);
                    		e.printStackTrace();
                    	}
                    }
                }
            };

            /*
             * This needs to be done in a thread for three reasons:
             * 1) If NotificationListener.subscribe() connects to a Slide instance
             *    that is in the process of starting it will wait until the server
             *    has finished starting before it returns. If configuration is
             *    single-thread this prevents this Slide instance from starting
             *    until all other Slide instances have started. This means none
             *    of them can start if they're all waiting for each other. Simple
             *    test case of this is a cluster of one instance. It never starts.
             * 2) Allows for renewing of subscriptions.
             * 3) Allows for retrying failed subscriptions. This will happen if
             *    a server is down and NotificationListener.subscribe() can't
             *    reach it.
             */
            Thread t = new Thread(new Runnable() {
            	
            	private boolean success;
            	
                public void run() {
                	success = true;
                    listener = new NotificationListener(host, port, repositoryHost, repositoryPort, protocol, credentials,
                        repositoryDomain, pollInterval, udp);

                    success = listener.subscribe("Update", uri, depth, lifetime, notificationDelay, contentSubscriber, credentials);
                    success = listener.subscribe("Update/newmember", uri, depth, lifetime, notificationDelay, structureSubscriber, credentials);
                    success = listener.subscribe("Delete", uri, depth, lifetime, notificationDelay, structureSubscriber, credentials);
                    success = listener.subscribe("Move", uri, depth, lifetime, notificationDelay, structureSubscriber, credentials);
                    
                    if ( !success ) {
						// try again quickly
                    	try {
							Thread.sleep(10000);
						} catch (InterruptedException e) {
							// ignore
						}
                    } else {
						// try again before the subscriptions expire
                    	try {
							Thread.sleep(lifetime*1000-60);
						} catch (InterruptedException e) {
							// ignore
						}
                    }
                }
            });
            t.setDaemon(true);
            t.setName("ClusterCacheRefresher");
            t.start();
        }
    }
    
    /**
     * Removes the first segment of a uri. "/slide/files/foo" becomes
     * "/files/foo".
     * 
     * @param uri the uri to strip
     * @return the stipped uri
     */
    private String stripUri(String uri) {
        // FIXME: if this is intended to remove the servlet path this will 
        // NOT work if the servlet is not default-servlet or is the root servlet
    	if ( uri.indexOf("/") == 0 ) {
    		uri = uri.substring(1);
    	}
    	if ( uri.indexOf("/") > -1 ) {
    		uri = uri.substring(uri.indexOf("/"));
    	}
    	return uri;
    }
}