/*
 * $Header: /home/cvspublic/jakarta-slide/src/webdav/server/org/apache/slide/webdav/event/NotificationTrigger.java,v 1.15 2005/03/22 17:27:47 masonjm Exp $
 * $Revision: 1.15 $
 * $Date: 2005/03/22 17:27:47 $
 *
 * ====================================================================
 *
 * Copyright 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.webdav.event;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.httpclient.HttpConnection;
import org.apache.commons.httpclient.HttpState;
import org.apache.commons.xmlio.in.ConversionHelpers;
import org.apache.commons.xmlio.in.DefaultSimpleImportHandler;
import org.apache.commons.xmlio.in.SimpleImporter;
import org.apache.commons.xmlio.in.SimplePath;
import org.apache.commons.xmlio.out.XMLEncode;
import org.apache.commons.xmlio.out.XMLOutputStreamWriter;
import org.apache.commons.xmlio.out.XMLWriter;
import org.apache.slide.common.Domain;
import org.apache.slide.event.ContentEvent;
import org.apache.slide.event.EventCollection;
import org.apache.slide.event.EventCollectionFilter;
import org.apache.slide.event.EventCollectionListener;
import org.apache.slide.event.ResourceEvent;
import org.apache.slide.event.VetoException;
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.slide.webdav.util.NotificationConstants;
import org.xml.sax.InputSource;
import org.xml.sax.helpers.AttributesImpl;

/**
 * @version $Revision: 1.15 $
 */
public class NotificationTrigger implements NotificationConstants, EventCollectionListener, Configurable {
    protected static final String LOG_CHANNEL = NotificationTrigger.class.getName();
    private final static String A_INCLUDE_EVENTS = "include-events";
    private final static String A_FILENAME = "filename";

    private final static String TCP_PROTOCOL = "http://";
    private final static String UDP_PROTOCOL = "httpu://";

    private final static String E_SUBSCRIPTIONS = "subscriptions";
    private final static String E_SUBSCRIPTION = "subscription";
    private final static String A_ID = "id";
    private final static String E_URI = "uri";
    private final static String E_DEPTH = "depth";
    private final static String E_NOTIFICATION_DELAY = "notification-delay";
    private final static String E_NOTIFICATION_TYPE = "notification-type";
    private final static String E_CALLBACK = "callback";
    private final static String E_SUBSCRIPTION_END = "subscription-end";
    
    protected static final Timer timer = new Timer(true);

    protected List subscribers = new ArrayList();
    protected int subscriberId = 0;
    protected boolean includeEvents = false;
    protected DatagramSocket socket;
    protected String filename = null;

    private static NotificationTrigger notificationTrigger = new NotificationTrigger();

    private NotificationTrigger() {
        Domain.log("Creating notification trigger", LOG_CHANNEL, Logger.INFO);
        try {
            socket = new DatagramSocket();
        } catch ( SocketException exception ) {
            Domain.log("Server socket creation failed, no UDP notifications available", LOG_CHANNEL, Logger.ERROR);
            socket = null;
        }
    }

    public static NotificationTrigger getInstance() {
        return notificationTrigger;
    }

    public int addSubscriber(Subscriber subscriber) {
        Domain.log("Adding subscriber", LOG_CHANNEL, Logger.INFO);
        subscriberId++;
        subscriber.setId(subscriberId);
        subscribers.add(subscriber);
        refreshSubscriber(subscriber, true);
        return subscriberId;
    }

    public boolean removeSubscriber(Subscriber subscriber) {
       Domain.log("Removing subscriber with ID: "+subscriber.getId(), LOG_CHANNEL, Logger.INFO);
       subscriber.getLifetime().cancel();
       saveSubscribers();
       return subscribers.remove(subscriber);
    }

    public void refreshSubscriber(final Subscriber subscriber, boolean persist) {
        TimerTask lifetimeTask = subscriber.getLifetime();
        if ( lifetimeTask != null ) lifetimeTask.cancel();
        if ( subscriber.getSubscriptionLifetime() > 0 ) {
            Domain.log("Refreshing subscriber with ID: "+subscriber.getId(), LOG_CHANNEL, Logger.INFO);
            TimerTask lifetime = new TimerTask() {
                public void run() {
                    Domain.log("Removing subscriber with ID: "+subscriber.getId(), LOG_CHANNEL, Logger.INFO);
                    refreshSubscriber(subscriber, true);
                }
            };
            subscriber.setLifetime(lifetime);
            timer.schedule(lifetime, subscriber.getSubscriptionLifetime()*1000);
        }
        if ( persist ) saveSubscribers();
    }

    public List getSubscribers() {
        return subscribers;
    }

    public Subscriber getSubscriber(int id) {
        for ( Iterator i = subscribers.iterator(); i.hasNext(); ) {
            Subscriber subsciber = (Subscriber)i.next();
            if ( subsciber.getId() == id ) {
                return subsciber;
            }
        }
        return null;
    }

    public void vetoableCollected(EventCollection collection) throws VetoException {
    }

    public void collected(EventCollection collection) {
        notifySubscribers(collection);
    }

    private void notifySubscribers(EventCollection collection) {
        Map subscriberEnumerations = new HashMap();
        List matchingSubscribers = new ArrayList();

        // get subscribers with matching notification types
        // (and remember events)
        ContentEvent[] update = EventCollectionFilter.getChangedContents(collection);
        for ( int i = 0; i < update.length; i++ ) {
            matchingSubscribers.addAll(getSubscribers(Subscriber.UPDATE, update[i]));
        }
        ContentEvent[] create = EventCollectionFilter.getCreatedContents(collection);
        for ( int i = 0; i < create.length; i++ ) {
            matchingSubscribers.addAll(getSubscribers(Subscriber.NEW_MEMBER, create[i]));
        }
        ContentEvent[] delete = EventCollectionFilter.getRemovedContents(collection);
        for ( int i = 0; i < delete.length; i++ ) {
            matchingSubscribers.addAll(getSubscribers(Subscriber.DELETE, delete[i]));
        }
        // FIXME: Add methods for MOVE, and NEW_MAIL (??) to get full exchange notification compliance
        
        // notifiy subscribers
        for ( Iterator i = matchingSubscribers.iterator(); i.hasNext(); ) {
           final Subscriber subscriber = (Subscriber)i.next();
           
           // skip subscribers that has no callback (we can't notify them)
           if (!subscriber.hasCallback()) continue;
           
           if ( subscriber.getNotificationDelay() == 0 ) {
              // send notification without delay
              List idList = (List)subscriberEnumerations.get(subscriber.getCallback());
              if ( idList == null ) {
                 idList = new ArrayList();
                 subscriberEnumerations.put(subscriber.getCallback(), idList);
              }
              Integer subscriberId = new Integer(subscriber.getId());
              if ( !idList.contains(subscriberId) ) {
                 idList.add(subscriberId);
              }
           } else {
              // send delayed notification
              TimerTask notifyTask = subscriber.getNotify();
              if ( notifyTask == null ) {
                 Domain.log("Starting notification delay: "+subscriber.getNotificationDelay(), 
                       LOG_CHANNEL, Logger.INFO);
                 notifyTask = new TimerTask() {
                    public void run() {
                       notifySubscriber(subscriber.getCallback(), 
                             String.valueOf(subscriber.getId()));
                       subscriber.setNotify(null);
                    }
                 };
                 subscriber.setNotify(notifyTask);
                 timer.schedule(notifyTask, subscriber.getNotificationDelay()*1000);
              }
           }
        }
        for ( Iterator i = subscriberEnumerations.entrySet().iterator(); i.hasNext(); ) {
            Map.Entry entry = (Map.Entry)i.next();
            String callBack = (String)entry.getKey();
            List idList = (List)entry.getValue();
            StringBuffer subscriberBuffer = new StringBuffer(128);
            boolean firstSubscriber = true;
            for ( Iterator j = idList.iterator(); j.hasNext(); ) {
                Integer id = (Integer)j.next();
                if ( !firstSubscriber ) {
                    subscriberBuffer.append(", ");
                }
                firstSubscriber = false;
                subscriberBuffer.append(id);
            }
            if ( !firstSubscriber ) {
                notifySubscriber(callBack, subscriberBuffer.toString());
            }
        }
    }

    protected void notifySubscriber(String callback, String subscribers) {
        if ( callback.startsWith(TCP_PROTOCOL) ) {
            Domain.log("Notify subscribers with adress='"+callback+"' via TCP with id's "+subscribers, LOG_CHANNEL, Logger.INFO);
            NotifyMethod notifyMethod = new NotifyMethod(callback.toString());
            notifyMethod.addRequestHeader(H_SUBSCRIPTION_ID_RESPONSE, subscribers);
            try {
                URL url = new URL(callback);
                notifyMethod.execute(
                      new HttpState(), new HttpConnection(url.getHost(), 
                            url.getPort()!=-1 ? url.getPort() : 80));
            } catch (IOException e) {
                Domain.log("Notification of subscriber '"+callback.toString()+"' failed!");
            }
        } else if ( callback.startsWith(UDP_PROTOCOL) && socket != null ) {
            Domain.log("Notify subscribers with adress='"+callback+"' via UDP with id's "+subscribers+"\n", LOG_CHANNEL, Logger.INFO);
            try {
                URL url = new URL(TCP_PROTOCOL+callback.substring(UDP_PROTOCOL.length()));
                String notification = "NOTIFY "+callback+" HTTP/1.1\nSubscription-id: "+subscribers;
                byte[] buf = notification.getBytes();
                InetAddress address = InetAddress.getByName(url.getHost());
                DatagramPacket packet = new DatagramPacket(
                      buf, buf.length, address, url.getPort()!=-1 ? url.getPort() : 80);
                socket.send(packet);
            } catch (IOException e) {
                Domain.log("Notification of subscriber '"+callback.toString()+"' failed!", LOG_CHANNEL, Logger.ERROR);
            }
        }
    }

    private List getSubscribers(String type, ResourceEvent event) {
        List matchingSubscribers = new ArrayList();
        for ( Iterator i = subscribers.iterator(); i.hasNext(); ) {
            Subscriber subscriber = (Subscriber)i.next();
            if ( subscriber.matches(type, event)) {
                matchingSubscribers.add(subscriber);
            	// remember this event for later poll method call
            	subscriber.addEvent(event);
            }
        }
        return matchingSubscribers;
    }

    public void configure(Configuration configuration) throws ConfigurationException {
        Configuration notification = configuration.getConfiguration("notification");
        includeEvents = notification.getAttributeAsBoolean(A_INCLUDE_EVENTS, false);
        Configuration persistSubscriptions = configuration.getConfiguration("persist-subscriptions");
        if  ( persistSubscriptions != null ) {
        	filename = persistSubscriptions.getAttribute(A_FILENAME);
        }
        loadSubscribers();
    }
    
    private void loadSubscribers() {
    	if ( filename != null ) {
    		synchronized ( subscribers ) {
    			File file = new File(filename);
    			if ( file.exists() ) {
    				try {
    					FileInputStream inputStream = new FileInputStream(filename);
    					SimpleImporter importer = new SimpleImporter();
    					importer.addSimpleImportHandler(new DefaultSimpleImportHandler() {
    						String callback, notificationType, uri;
    						int depth, notificationDelay, subscriptionLifetime, id;
    						List events = new ArrayList();
    						
    						public void startElement(SimplePath path, String name, AttributesImpl attributes, String leadingCDdata) {
    							if ( path.matches(E_SUBSCRIPTION) ) {
    								id = ConversionHelpers.getInt(attributes.getValue(A_ID));
    							} else if ( path.matches(E_URI) ) {
    								uri = leadingCDdata;
    							} else if ( path.matches(E_DEPTH) ) {
    								depth = Integer.valueOf(leadingCDdata).intValue();
    							} else if ( path.matches(E_CALLBACK) ) {
    								callback = leadingCDdata;
    							} else if ( path.matches(E_NOTIFICATION_DELAY) ) {
    								notificationDelay = Integer.valueOf(leadingCDdata).intValue();
    							} else if ( path.matches(E_NOTIFICATION_TYPE) ) {
    								notificationType = leadingCDdata;
    							} else if ( path.matches(E_SUBSCRIPTION_END) ) {
    								subscriptionLifetime = (int)(Long.valueOf(leadingCDdata).longValue() - System.currentTimeMillis());
    							}
    						}
    						
    						public void endElement(SimplePath path, String name) {
    							if ( path.matches(E_SUBSCRIPTION) ) {
    								Subscriber subscriber = new Subscriber(uri, callback, notificationType, notificationDelay, subscriptionLifetime, depth); 
    								subscribers.add(subscriber);
    								refreshSubscriber(subscriber, false);
    							}
    						}
    					});
    					importer.parse(new InputSource(inputStream));
    				} catch (Exception e) {
    					Domain.log("Exception while restoring subscriptions. Skipping...");
    				}				
    			}
    		}
    	}
    }
    
    private void saveSubscribers() {
    	if ( filename != null ) {
    		synchronized ( subscribers ) {
    			try {
    				FileOutputStream outputStream = new FileOutputStream(filename);
    				XMLOutputStreamWriter writer = new XMLOutputStreamWriter(outputStream);
    				writer.writeXMLDeclaration();
    				writer.writeStartTag(XMLWriter.createStartTag(E_SUBSCRIPTIONS));
    				for ( Iterator i = subscribers.iterator(); i.hasNext(); ) {
    					Subscriber subscriber = (Subscriber)i.next();
    					writer.writeStartTag(XMLWriter.createStartTag(E_SUBSCRIPTION, new String[][] { 
    							{ A_ID, String.valueOf(subscriber.getId()) } }));
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_URI), XMLEncode.xmlEncodeText(subscriber.getUri()), XMLWriter.createEndTag(E_URI)); 
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_DEPTH), String.valueOf(subscriber.getDepth()), XMLWriter.createEndTag(E_DEPTH)); 
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_CALLBACK), XMLEncode.xmlEncodeText(subscriber.getCallback()), XMLWriter.createEndTag(E_CALLBACK)); 
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_NOTIFICATION_TYPE), XMLEncode.xmlEncodeText(subscriber.getNotificationType()), XMLWriter.createEndTag(E_NOTIFICATION_TYPE)); 
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_NOTIFICATION_DELAY), String.valueOf(subscriber.getNotificationDelay()), XMLWriter.createEndTag(E_NOTIFICATION_DELAY)); 
    					writer.writeElementWithPCData(XMLWriter.createStartTag(E_SUBSCRIPTION_END), String.valueOf(subscriber.getSubscriptionEnd()), XMLWriter.createEndTag(E_SUBSCRIPTION_END)); 
    					writer.writeEndTag(XMLWriter.createEndTag(E_SUBSCRIPTION)); 
    				}
    				writer.writeEndTag(XMLWriter.createEndTag(E_SUBSCRIPTIONS));
    				outputStream.close();
    			} catch ( Exception e) {
    				Domain.log(e);
    			}
    		}
    	}
    }
}