/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.sling.event.impl;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;

import javax.jcr.Item;
import javax.jcr.Node;
import javax.jcr.NodeIterator;
import javax.jcr.RepositoryException;
import javax.jcr.Session;
import javax.jcr.Value;
import javax.jcr.lock.Lock;
import javax.jcr.lock.LockException;
import javax.jcr.observation.EventIterator;
import javax.jcr.query.Query;
import javax.jcr.query.QueryManager;

import org.apache.sling.commons.scheduler.Job;
import org.apache.sling.commons.scheduler.JobContext;
import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.event.EventUtil;
import org.apache.sling.event.TimedEventStatusProvider;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;


/**
 * An event handler for timed events.
 *
 * @scr.component metatype="no"
 * @scr.service interface="TimedEventStatusProvider"
 * @scr.property name="event.topics" refValues="EventUtil.TOPIC_TIMED_EVENT"
 *               values.updated="org/osgi/framework/BundleEvent/UPDATED"
 *               values.started="org/osgi/framework/BundleEvent/STARTED"
 * @scr.property name="repository.path" value="/var/eventing/timed-jobs"
 */
public class TimedJobHandler
    extends AbstractRepositoryEventHandler
    implements Job, TimedEventStatusProvider {

    protected static final String JOB_TOPIC = "topic";

    protected static final String JOB_CONFIG = "config";

    protected static final String JOB_SCHEDULE_INFO = "info";

    /** @scr.reference */
    protected Scheduler scheduler;

    /** Unloaded events. */
    protected Set<String>unloadedEvents = new HashSet<String>();

    /** Sync lock */
    private final Object writeLock = new Object();

    /**
     * @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#startWriterSession()
     */
    protected void startWriterSession() throws RepositoryException {
        super.startWriterSession();
        // load timed events from repository
        this.loadEvents();
        this.writerSession.getWorkspace().getObservationManager()
            .addEventListener(this, javax.jcr.observation.Event.PROPERTY_CHANGED|javax.jcr.observation.Event.PROPERTY_REMOVED, this.repositoryPath, true, null, null, true);
    }

    /**
     * @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#processWriteQueue()
     */
    protected void processWriteQueue() {
        while ( this.running ) {
            Event event = null;
            try {
                event = this.writeQueue.take();
            } catch (InterruptedException e) {
                // we ignore this
                this.ignoreException(e);
            }
            if ( this.running && event != null ) {
                ScheduleInfo scheduleInfo = null;
                try {
                    scheduleInfo = new ScheduleInfo(event);
                } catch (IllegalArgumentException iae) {
                    this.logger.error(iae.getMessage());
                }
                if ( scheduleInfo != null ) {
                    final EventInfo info = new EventInfo();
                    info.event = event;

                    // write event and update path
                    // if something went wrong we get the node path and reschedule
                    synchronized ( this.writeLock ) {
                        info.nodePath = this.persistEvent(info.event, scheduleInfo);
                    }
                    if ( info.nodePath != null ) {
                        try {
                            this.queue.put(info);
                        } catch (InterruptedException e) {
                            // this should never happen, so we ignore it
                            this.ignoreException(e);
                        }
                    }
                }
            }
        }
    }

    /**
     * @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#runInBackground()
     */
    protected void runInBackground() {
        while ( this.running ) {
            // so let's wait/get the next info from the queue
            EventInfo info = null;
            try {
                info = this.queue.take();
            } catch (InterruptedException e) {
                // we ignore this
                this.ignoreException(e);
            }
            if ( info != null && this.running ) {
                synchronized ( this.writeLock ) {
                    ScheduleInfo scheduleInfo = null;
                    try {
                        scheduleInfo = new ScheduleInfo(info.event);
                    } catch (IllegalArgumentException iae) {
                        this.logger.error(iae.getMessage());
                    }
                    if ( scheduleInfo != null ) {
                        try {
                            this.writerSession.refresh(true);
                            if ( this.writerSession.itemExists(info.nodePath) ) {
                                final Node eventNode = (Node) this.writerSession.getItem(info.nodePath);
                                if ( !eventNode.isLocked() ) {
                                    // lock node
                                    Lock lock = null;
                                    try {
                                        lock = eventNode.lock(false, true);
                                    } catch (RepositoryException re) {
                                        // lock failed which means that the node is locked by someone else, so we don't have to requeue
                                    }
                                    if ( lock != null ) {
                                        // if something went wrong, we reschedule
                                        if ( !this.processEvent(info.event, scheduleInfo) ) {
                                            try {
                                                this.queue.put(info);
                                            } catch (InterruptedException e) {
                                                // this should never happen, so we ignore it
                                                this.ignoreException(e);
                                            }
                                        }
                                    }
                                }
                            }
                        } catch (RepositoryException e) {
                            // ignore
                            this.ignoreException(e);
                        }
                    }
                }
            }
        }
    }

    protected String persistEvent(final Event event, final ScheduleInfo scheduleInfo) {
        try {
            // get parent node
            final Node parentNode = this.ensureRepositoryPath();
            final String nodeName = scheduleInfo.jobId;
            // is there already a node?
            final Node foundNode = parentNode.hasNode(nodeName) ? parentNode.getNode(nodeName) : null;
            Lock lock = null;
            if ( scheduleInfo.isStopEvent() ) {
                // if this is a stop event, we should remove the node from the repository
                // if there is no node someone else was faster and we can ignore this
                if ( foundNode != null ) {
                    try {
                        foundNode.remove();
                        parentNode.save();
                    } catch (LockException le) {
                        // if someone else has the lock this is fine
                    }
                }
                // stop the scheduler
                processEvent(event, scheduleInfo);
            } else {
                // if there is already a node, it means we must handle an update
                if ( foundNode != null ) {
                    try {
                        foundNode.remove();
                        parentNode.save();
                    } catch (LockException le) {
                        // if someone else has the lock this is fine
                    }
                    // create a stop event
                    processEvent(event, scheduleInfo.getStopInfo());
                }
                // we only write the event if this is a local one
                if ( EventUtil.isLocal(event) ) {

                    // write event to repository, lock it and schedule the event
                    final Node eventNode = writeEvent(event, nodeName);
                    lock = eventNode.lock(false, true);
                }
            }

            if ( lock != null ) {
                // if something went wrong, we reschedule
                if ( !this.processEvent(event, scheduleInfo) ) {
                    final String path = lock.getNode().getPath();
                    lock.getNode().unlock();
                    return path;
                }
            }
        } catch (RepositoryException re ) {
            // something went wrong, so let's log it
            this.logger.error("Exception during writing new job to repository.", re);
        }
        return null;
    }

    /**
     * Process the event.
     * If a scheduler is available, a job is scheduled or stopped.
     * @param event The incomming event.
     * @return
     */
    protected boolean processEvent(final Event event, final ScheduleInfo scheduleInfo) {
        final Scheduler localScheduler = this.scheduler;
        if ( localScheduler != null ) {
            // is this a stop event?
            if ( scheduleInfo.isStopEvent() ) {
                if ( this.logger.isDebugEnabled() ) {
                    this.logger.debug("Stopping timed event " + event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_TOPIC) + "(" + scheduleInfo.jobId + ")");
                }
                try {
                    localScheduler.removeJob(scheduleInfo.jobId);
                } catch (NoSuchElementException nsee) {
                    // this can happen if the job is scheduled on another node
                    // so we can just ignore this
                }
                return true;
            }
            // we ignore remote job events
            if ( !EventUtil.isLocal(event) ) {
                return true;
            }

            // Create configuration for scheduled job
            final Map<String, Serializable> config = new HashMap<String, Serializable>();
            // copy properties
            final Hashtable<String, Object> properties = new Hashtable<String, Object>();
            config.put(JOB_TOPIC, (String)event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_TOPIC));
            final String[] names = event.getPropertyNames();
            if ( names != null ) {
                for(int i=0; i<names.length; i++) {
                    properties.put(names[i], event.getProperty(names[i]));
                }
            }
            config.put(JOB_CONFIG, properties);
            config.put(JOB_SCHEDULE_INFO, scheduleInfo);

            try {
                if ( scheduleInfo.expression != null ) {
                    if ( this.logger.isDebugEnabled() ) {
                        this.logger.debug("Adding timed event " + config.get(JOB_TOPIC) + "(" + scheduleInfo.jobId + ")" + " with cron expression " + scheduleInfo.expression);
                    }
                    localScheduler.addJob(scheduleInfo.jobId, this, config, scheduleInfo.expression, false);
                } else if ( scheduleInfo.period != null ) {
                    if ( this.logger.isDebugEnabled() ) {
                        this.logger.debug("Adding timed event " + config.get(JOB_TOPIC) + "(" + scheduleInfo.jobId + ")" + " with period " + scheduleInfo.period);
                    }
                    localScheduler.addPeriodicJob(scheduleInfo.jobId, this, config, scheduleInfo.period, false);
                } else {
                    // then it must be date
                    if ( this.logger.isDebugEnabled() ) {
                        this.logger.debug("Adding timed event " + config.get(JOB_TOPIC) + "(" + scheduleInfo.jobId + ")" + " with date " + scheduleInfo.date);
                    }
                    localScheduler.fireJobAt(scheduleInfo.jobId, this, config, scheduleInfo.date);
                }
                return true;
            } catch (Exception e) {
                this.ignoreException(e);
            }
        } else {
            this.logger.error("No scheduler available to start timed event " + event);
        }
        return false;
    }

    /**
     * @see javax.jcr.observation.EventListener#onEvent(javax.jcr.observation.EventIterator)
     */
    public void onEvent(EventIterator iter) {
        // we create an own session here
        Session s = null;
        try {
            s = this.createSession();
            while ( iter.hasNext() ) {
                final javax.jcr.observation.Event event = iter.nextEvent();
                if ( event.getType() == javax.jcr.observation.Event.PROPERTY_CHANGED
                    || event.getType() == javax.jcr.observation.Event.PROPERTY_REMOVED) {

                    final String propPath = event.getPath();
                    int pos = propPath.lastIndexOf('/');
                    final String nodePath = propPath.substring(0, pos);
                    final String propertyName = propPath.substring(pos+1);
                    // we are only interested in unlocks
                    if ( "jcr:lockOwner".equals(propertyName) ) {
                        try {
                            final Node eventNode = (Node) s.getItem(nodePath);
                            if ( !eventNode.isLocked() ) {
                                try {
                                    final EventInfo info = new EventInfo();
                                    info.event = this.readEvent(eventNode);
                                    info.nodePath =nodePath;
                                    try {
                                        this.queue.put(info);
                                    } catch (InterruptedException e) {
                                        // we ignore this exception as this should never occur
                                        this.ignoreException(e);
                                    }
                                } catch (ClassNotFoundException cnfe) {
                                    // add it to the unloaded set
                                    synchronized (unloadedEvents) {
                                        this.unloadedEvents.add(nodePath);
                                    }
                                    this.ignoreException(cnfe);
                                }
                            }
                        } catch (RepositoryException re) {
                            this.logger.error("Exception during jcr event processing.", re);
                        }
                    }
                }
            }
        } catch (RepositoryException re) {
            this.logger.error("Unable to create a session.", re);
        } finally {
            if ( s != null ) {
                s.logout();
            }
        }
    }

    /**
     * @see org.osgi.service.event.EventHandler#handleEvent(org.osgi.service.event.Event)
     */
    public void handleEvent(Event event) {
        if ( event.getTopic().equals(EventUtil.TOPIC_TIMED_EVENT) ) {
            // queue the event in order to respond quickly
            try {
                this.writeQueue.put(event);
            } catch (InterruptedException e) {
                // this should never happen
                this.ignoreException(e);
            }
        } else {
            // bundle event started or updated
            boolean doIt = false;
            synchronized ( this.unloadedEvents ) {
                if ( this.unloadedEvents.size() > 0 ) {
                    doIt = true;
                }
            }
            if ( doIt ) {
                final Runnable t = new Runnable() {

                    public void run() {
                        synchronized (unloadedEvents) {
                            Session s = null;
                            final Set<String> newUnloadedEvents = new HashSet<String>();
                            newUnloadedEvents.addAll(unloadedEvents);
                            try {
                                s = createSession();
                                for(String path : unloadedEvents ) {
                                    newUnloadedEvents.remove(path);
                                    try {
                                        if ( s.itemExists(path) ) {
                                            final Node eventNode = (Node) s.getItem(path);
                                            if ( !eventNode.isLocked() ) {
                                                try {
                                                    final EventInfo info = new EventInfo();
                                                    info.event = readEvent(eventNode);
                                                    info.nodePath = path;
                                                    try {
                                                        queue.put(info);
                                                    } catch (InterruptedException e) {
                                                        // we ignore this exception as this should never occur
                                                        ignoreException(e);
                                                    }
                                                } catch (ClassNotFoundException cnfe) {
                                                    newUnloadedEvents.add(path);
                                                    ignoreException(cnfe);
                                                }
                                            }
                                        }
                                    } catch (RepositoryException re) {
                                        // we ignore this and readd
                                        newUnloadedEvents.add(path);
                                        ignoreException(re);
                                    }
                                }
                            } catch (RepositoryException re) {
                                // unable to create session, so we try it again next time
                                ignoreException(re);
                            } finally {
                                if ( s != null ) {
                                    s.logout();
                                }
                                unloadedEvents.clear();
                                unloadedEvents.addAll(newUnloadedEvents);
                            }
                        }
                    }

                };
                this.threadPool.execute(t);
            }
        }
    }

    /**
     * @see org.apache.sling.commons.scheduler.Job#execute(org.apache.sling.commons.scheduler.JobContext)
     */
    public void execute(JobContext context) {
        final String topic = (String) context.getConfiguration().get(JOB_TOPIC);
        @SuppressWarnings("unchecked")
        final Dictionary<Object, Object> properties = (Dictionary<Object, Object>) context.getConfiguration().get(JOB_CONFIG);
        final EventAdmin ea = this.eventAdmin;
        if ( ea != null ) {
            try {
                ea.postEvent(new Event(topic, properties));
            } catch (IllegalArgumentException iae) {
                this.logger.error("Scheduled event has illegal topic: " + topic, iae);
            }
        } else {
            this.logger.warn("Unable to send timed event as no event admin service is available.");
        }
        final ScheduleInfo info = (ScheduleInfo) context.getConfiguration().get(JOB_SCHEDULE_INFO);
        // is this job scheduled for a specific date?
        if ( info.date != null ) {
            // we can remove it from the repository
            // we create an own session here
            Session s = null;
            try {
                s = this.createSession();
                if ( s.itemExists(this.repositoryPath) ) {
                    final Node parentNode = (Node)s.getItem(this.repositoryPath);
                    final String nodeName = info.jobId;
                    final Node eventNode = parentNode.hasNode(nodeName) ? parentNode.getNode(nodeName) : null;
                    if ( eventNode != null ) {
                        try {
                            eventNode.remove();
                            parentNode.save();
                        } catch (RepositoryException re) {
                            // we ignore the exception if removing fails
                            ignoreException(re);
                        }
                    }
                }
            } catch (RepositoryException re) {
                this.logger.error("Unable to create a session.", re);
            } finally {
                if ( s != null ) {
                    s.logout();
                }
            }
        }
    }

    /**
     * Load all active timed events from the repository.
     * @throws RepositoryException
     */
    protected void loadEvents() {
        try {
            final QueryManager qManager = this.writerSession.getWorkspace().getQueryManager();
            final StringBuffer buffer = new StringBuffer("/jcr:root");
            buffer.append(this.repositoryPath);
            buffer.append("//element(*, ");
            buffer.append(this.getEventNodeType());
            buffer.append(")");
            final Query q = qManager.createQuery(buffer.toString(), Query.XPATH);
            final NodeIterator result = q.execute().getNodes();
            while ( result.hasNext() ) {
                final Node eventNode = result.nextNode();
                if ( !eventNode.isLocked() ) {
                    final String nodePath = eventNode.getPath();
                    try {
                        final Event event = this.readEvent(eventNode);
                        final EventInfo info = new EventInfo();
                        info.event = event;
                        info.nodePath = nodePath;
                        try {
                            this.queue.put(info);
                        } catch (InterruptedException e) {
                            // we ignore this exception as this should never occur
                            this.ignoreException(e);
                        }
                    } catch (ClassNotFoundException cnfe) {
                        // add it to the unloaded set
                        synchronized (unloadedEvents) {
                            this.unloadedEvents.add(nodePath);
                        }
                        this.ignoreException(cnfe);
                    } catch (RepositoryException re) {
                        // if reading an event fails, we ignore this
                        this.ignoreException(re);
                    }
                }
            }
        } catch (RepositoryException re) {
            this.logger.error("Exception during initial loading of stored timed events.", re);
        }
    }

    /**
     * @see org.apache.sling.engine.event.impl.JobPersistenceHandler#addNodeProperties(javax.jcr.Node, org.osgi.service.event.Event)
     */
    protected void addNodeProperties(Node eventNode, Event event)
    throws RepositoryException {
        super.addNodeProperties(eventNode, event);
        eventNode.setProperty(EventHelper.NODE_PROPERTY_TOPIC, (String)event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_TOPIC));
        final ScheduleInfo info = new ScheduleInfo(event);
        if ( info.date != null ) {
            final Calendar c = Calendar.getInstance();
            c.setTime(info.date);
            eventNode.setProperty(EventHelper.NODE_PROPERTY_TE_DATE, c);
        }
        if ( info.expression != null ) {
            eventNode.setProperty(EventHelper.NODE_PROPERTY_TE_EXPRESSION, info.expression);
        }
        if ( info.period != null ) {
            eventNode.setProperty(EventHelper.NODE_PROPERTY_TE_PERIOD, info.period.longValue());
        }
    }

    /**
     * @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#getEventNodeType()
     */
    protected String getEventNodeType() {
        return EventHelper.TIMED_EVENT_NODE_TYPE;
    }

    protected static final class ScheduleInfo implements Serializable {

        public final String expression;
        public final Long   period;
        public final Date   date;
        public final String jobId;

        public ScheduleInfo(final Event event)
        throws IllegalArgumentException {
            // let's see if a schedule information is specified or if the job should be stopped
            this.expression = (String) event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_SCHEDULE);
            this.period = (Long) event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_PERIOD);
            this.date = (Date) event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_DATE);
            int count = 0;
            if ( this.expression != null) {
                count++;
            }
            if ( this.period != null ) {
                count++;
            }
            if ( this.date != null ) {
                count++;
            }
            if ( count > 1 ) {
                throw new IllegalArgumentException("Only one configuration property from " + EventUtil.PROPERTY_TIMED_EVENT_SCHEDULE +
                                      ", " + EventUtil.PROPERTY_TIMED_EVENT_PERIOD +
                                      ", or " + EventUtil.PROPERTY_TIMED_EVENT_DATE + " should be used.");
            }
            // we create a job id consisting of the real event topic and an (optional) id
            // if the event contains a timed event id or a job id we'll append that to the name
            String topic = (String)event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_TOPIC);
            if ( topic == null ) {
                throw new IllegalArgumentException("Timed event does not contain required property " + EventUtil.PROPERTY_TIMED_EVENT_TOPIC);
            }
            String id = (String)event.getProperty(EventUtil.PROPERTY_TIMED_EVENT_ID);
            String jId = (String)event.getProperty(EventUtil.PROPERTY_JOB_ID);

            //this.jobId = getJobId(topic, id, jId);
            this.jobId = getJobId(topic, id, jId);
        }

        private ScheduleInfo(String jobId) {
            this.expression = null;
            this.period = null;
            this.date = null;
            this.jobId = jobId;
        }

        public ScheduleInfo getStopInfo() {
            return new ScheduleInfo(this.jobId);
        }

        public boolean isStopEvent() {
            return this.expression == null && this.period == null && this.date == null;
        }

        public static String getJobId(String topic, String timedEventId, String jobId) {
            return topic.replace('/', '.') + "/TimedEvent " + (timedEventId != null ? EventHelper.filter(timedEventId) : "") + '_' + (jobId != null ? EventHelper.filter(jobId) : "");
        }
    }

    /**
     * @see org.apache.sling.event.TimedEventStatusProvider#getScheduledEvent(java.lang.String, java.lang.String, java.lang.String)
     */
    public Event getScheduledEvent(String topic, String eventId, String jobId) {
        Session s = null;
        try {
            s = this.createSession();
            if ( s.itemExists(this.repositoryPath) ) {
                final Node parentNode = (Node)s.getItem(this.repositoryPath);
                final String nodeName = ScheduleInfo.getJobId(topic, eventId, jobId);
                final Node eventNode = parentNode.hasNode(nodeName) ? parentNode.getNode(nodeName) : null;
                if ( eventNode != null ) {
                    return this.readEvent(eventNode);
                }
            }
        } catch (RepositoryException re) {
            this.logger.error("Unable to create a session.", re);
        } catch (ClassNotFoundException e) {
            this.ignoreException(e);
        } finally {
            if ( s != null ) {
                s.logout();
            }
        }
        return null;
    }

    /**
     * @see org.apache.sling.event.TimedEventStatusProvider#getScheduledEvents(java.lang.String, java.util.Map...)
     */
    public Collection<Event> getScheduledEvents(String topic, Map<String, Object>... filterProps) {
        // we create a new session
        Session s = null;
        final List<Event> jobs = new ArrayList<Event>();
        try {
            s = this.createSession();
            final QueryManager qManager = s.getWorkspace().getQueryManager();
            final StringBuffer buffer = new StringBuffer("/jcr:root");
            buffer.append(this.repositoryPath);
            if ( topic != null ) {
                buffer.append('/');
                buffer.append(topic.replace('/', '.'));
            }
            buffer.append("//element(*, ");
            buffer.append(this.getEventNodeType());
            buffer.append(")");
            if ( filterProps != null && filterProps.length > 0 ) {
                buffer.append(" [");
                int index = 0;
                for (Map<String,Object> template : filterProps) {
                    if ( index > 0 ) {
                        buffer.append(" or ");
                    }
                    buffer.append('(');
                    final Iterator<Map.Entry<String, Object>> i = template.entrySet().iterator();
                    boolean first = true;
                    while ( i.hasNext() ) {
                        final Map.Entry<String, Object> current = i.next();
                        // check prop name first
                        final String propName = EventUtil.getNodePropertyName(current.getKey());
                        if ( propName != null ) {
                            // check value
                            final Value value = EventUtil.getNodePropertyValue(s.getValueFactory(), current.getValue());
                            if ( value != null ) {
                                if ( first ) {
                                    first = false;
                                    buffer.append('@');
                                } else {
                                    buffer.append(" and @");
                                }
                                buffer.append(propName);
                                buffer.append(" = '");
                                buffer.append(current.getValue());
                                buffer.append("'");
                            }
                        }
                    }
                    buffer.append(')');
                    index++;
                }
                buffer.append(']');
            }
            final String queryString = buffer.toString();
            logger.debug("Executing job query {}.", queryString);

            final Query q = qManager.createQuery(queryString, Query.XPATH);
            final NodeIterator iter = q.execute().getNodes();
            while ( iter.hasNext() ) {
                final Node eventNode = iter.nextNode();
                try {
                    final Event event = this.readEvent(eventNode);
                    jobs.add(event);
                } catch (ClassNotFoundException cnfe) {
                    // in the case of a class not found exception we just ignore the exception
                    this.ignoreException(cnfe);
                }
            }
        } catch (RepositoryException e) {
            // in the case of an error, we return an empty list
            this.ignoreException(e);
        } finally {
            if ( s != null) {
                s.logout();
            }
        }
        return jobs;
    }

    /**
     * @see org.apache.sling.event.TimedEventStatusProvider#cancelTimedEvent(java.lang.String)
     */
    public void cancelTimedEvent(String jobId) {
        synchronized ( this.writeLock ) {
            try {
                // is there a node?
                final Item foundNode = this.writerSession.itemExists(jobId) ? this.writerSession.getItem(jobId) : null;
                // we should remove the node from the repository
                // if there is no node someone else was faster and we can ignore this
                if ( foundNode != null ) {
                    final Node parentNode = foundNode.getParent();
                    try {
                        foundNode.remove();
                        parentNode.save();
                    } catch (LockException le) {
                        // if someone else has the lock this is fine
                    }
                }
            } catch ( RepositoryException re) {
                this.logger.error("Unable to cancel timed event: " + jobId, re);
            }
            // stop the scheduler
            if ( this.logger.isDebugEnabled() ) {
                this.logger.debug("Stopping timed event " + jobId);
            }
            final Scheduler localScheduler = this.scheduler;
            if ( localScheduler != null ) {
                try {
                    localScheduler.removeJob(jobId);
                } catch (NoSuchElementException nsee) {
                    // this can happen if the job is scheduled on another node
                    // so we can just ignore this
                }
            }
        }
    }
}