/*
* 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.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import javax.jcr.ItemExistsException;
import javax.jcr.Node;
import javax.jcr.NodeIterator;
import javax.jcr.RepositoryException;
import javax.jcr.Session;
import javax.jcr.observation.EventIterator;
import javax.jcr.query.Query;
import javax.jcr.query.QueryManager;
import org.apache.jackrabbit.util.ISO8601;
import org.apache.sling.commons.osgi.OsgiUtil;
import org.apache.sling.event.EventUtil;
import org.apache.sling.event.JobStatusProvider;
import org.osgi.framework.BundleEvent;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
/**
* An event handler handling special job events.
*
* @scr.component label="%job.events.name" description="%job.events.description"
* @scr.service interface="org.apache.sling.event.JobStatusProvider"
* @scr.property name="event.topics" refValues="EventUtil.TOPIC_JOB"
* values.updated="org/osgi/framework/BundleEvent/UPDATED"
* values.started="org/osgi/framework/BundleEvent/STARTED"
* private="true"
* @scr.property name="repository.path" value="/var/eventing/jobs" private="true"
* We schedule this event handler to run in the background and clean up
* obsolete events.
* @scr.service interface="java.lang.Runnable"
* @scr.property name="scheduler.period" value="600" type="Long"
* @scr.property name="scheduler.concurrent" value="false" type="Boolean" private="true"
*/
public class JobEventHandler
extends AbstractRepositoryEventHandler
implements EventUtil.JobStatusNotifier, JobStatusProvider, Runnable {
/** The topic prefix for bundle events. */
protected static final String BUNDLE_EVENT_PREFIX = BundleEvent.class.getName().replace('.', '/') + '/';
/** A map for keeping track of currently processed job topics. */
protected final Map<String, Boolean> processingMap = new HashMap<String, Boolean>();
/** Default sleep time. */
protected static final long DEFAULT_SLEEP_TIME = 20;
/** @scr.property valueRef="DEFAULT_SLEEP_TIME" */
protected static final String CONFIG_PROPERTY_SLEEP_TIME = "sleep.time";
/** Default number of job retries. */
protected static final int DEFAULT_MAX_JOB_RETRIES = 10;
/** @scr.property valueRef="DEFAULT_MAX_JOB_RETRIES" */
protected static final String CONFIG_PROPERTY_MAX_JOB_RETRIES = "max.job.retries";
/** We check every 20 secs by default. */
protected long sleepTime;
/** How often should a job be retried by default. */
protected int maxJobRetries;
/** Background session. */
protected Session backgroundSession;
/** Unloaded jobs. */
protected Set<String>unloadedJobs = new HashSet<String>();
/** List of deleted jobs. */
protected Set<String>deletedJobs = new HashSet<String>();
/** Default clean up time is 10 minutes. */
protected static final int DEFAULT_CLEANUP_PERIOD = 10;
/** @scr.property valueRef="DEFAULT_CLEANUP_PERIOD" type="Integer" */
protected static final String CONFIG_PROPERTY_CLEANUP_PERIOD = "cleanup.period";
/** We remove everything which is older than 5 min by default. */
protected int cleanupPeriod = DEFAULT_CLEANUP_PERIOD;
/**
* Activate this component.
* @param context
* @throws RepositoryException
*/
protected void activate(final ComponentContext context)
throws Exception {
@SuppressWarnings("unchecked")
final Dictionary<String, Object> props = context.getProperties();
this.cleanupPeriod = OsgiUtil.toInteger(props.get(CONFIG_PROPERTY_CLEANUP_PERIOD), DEFAULT_CLEANUP_PERIOD);
this.sleepTime = OsgiUtil.toLong(props.get(CONFIG_PROPERTY_SLEEP_TIME), DEFAULT_SLEEP_TIME);
this.maxJobRetries = OsgiUtil.toInteger(props.get(CONFIG_PROPERTY_MAX_JOB_RETRIES), DEFAULT_MAX_JOB_RETRIES);
super.activate(context);
}
/**
* @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#deactivate(org.osgi.service.component.ComponentContext)
*/
protected void deactivate(final ComponentContext context) {
super.deactivate(context);
if ( this.backgroundSession != null ) {
try {
this.backgroundSession.getWorkspace().getObservationManager().removeEventListener(this);
} catch (RepositoryException e) {
// we just ignore it
this.logger.warn("Unable to remove event listener.", e);
}
this.backgroundSession.logout();
this.backgroundSession = null;
}
}
/**
* Return the query string for the clean up.
*/
protected String getCleanUpQueryString() {
final Calendar deleteBefore = Calendar.getInstance();
deleteBefore.add(Calendar.MINUTE, -this.cleanupPeriod);
final String dateString = ISO8601.format(deleteBefore);
final StringBuffer buffer = new StringBuffer("/jcr:root");
buffer.append(this.repositoryPath);
buffer.append("//element(*, ");
buffer.append(getEventNodeType());
buffer.append(")[@");
buffer.append(EventHelper.NODE_PROPERTY_FINISHED);
buffer.append(" < xs:dateTime('");
buffer.append(dateString);
buffer.append("')]");
return buffer.toString();
}
/**
* This method is invoked periodically.
* @see java.lang.Runnable#run()
*/
public void run() {
if ( this.cleanupPeriod > 0 ) {
this.logger.debug("Cleaning up repository, removing all finished jobs older than {} minutes.", this.cleanupPeriod);
final String queryString = this.getCleanUpQueryString();
// we create an own session for concurrency issues
Session s = null;
try {
s = this.createSession();
final Node parentNode = (Node)s.getItem(this.repositoryPath);
logger.debug("Executing query {}", queryString);
final Query q = s.getWorkspace().getQueryManager().createQuery(queryString, Query.XPATH);
final NodeIterator iter = q.execute().getNodes();
int count = 0;
while ( iter.hasNext() ) {
final Node eventNode = iter.nextNode();
eventNode.remove();
count++;
}
parentNode.save();
logger.debug("Removed {} entries from the repository.", count);
} catch (RepositoryException e) {
// in the case of an error, we just log this as a warning
this.logger.warn("Exception during repository cleanup.", e);
} finally {
if ( s != null ) {
s.logout();
}
}
}
}
/**
* @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#processWriteQueue()
*/
protected void processWriteQueue() {
while ( this.running ) {
// so let's wait/get the next job from the queue
Event event = null;
try {
event = this.writeQueue.take();
} catch (InterruptedException e) {
// we ignore this
this.ignoreException(e);
}
if ( event != null && this.running ) {
try {
this.writerSession.refresh(false);
} catch (RepositoryException re) {
// we just ignore this
this.ignoreException(re);
}
final EventInfo info = new EventInfo();
info.event = event;
final String nodeName = this.getNodeName(event);
// if the job has no job id, we can just write the job to the repo and don't
// need locking
final String jobId = (String)event.getProperty(EventUtil.PROPERTY_JOB_ID);
if ( jobId == null ) {
try {
final Node eventNode = this.writeEvent(event, nodeName);
info.nodePath = eventNode.getPath();
} catch (RepositoryException re ) {
// something went wrong, so let's log it
this.logger.error("Exception during writing new job '" + nodeName + "' to repository.", re);
}
} else {
try {
// let's first search for an existing node with the same id
final Node parentNode = (Node)this.writerSession.getItem(this.repositoryPath);
Node foundNode = null;
if ( parentNode.hasNode(nodeName) ) {
foundNode = parentNode.getNode(nodeName);
}
if ( foundNode != null ) {
// if the node is locked, someone else was quicker
// and we don't have to process this job
if ( !foundNode.isLocked() ) {
// node is already in repository, so if not finished we just use it
// otherwise it has already been processed
try {
if ( !foundNode.hasProperty(EventHelper.NODE_PROPERTY_FINISHED) ) {
info.nodePath = foundNode.getPath();
}
} catch (RepositoryException re) {
// if anything goes wrong, it means that (hopefully) someone
// else is processing this node
}
}
} else {
// We now write the event into the repository
try {
final Node eventNode = this.writeEvent(event, nodeName);
info.nodePath = eventNode.getPath();
} catch (ItemExistsException iee) {
// someone else did already write this node in the meantime
// nothing to do for us
}
}
} catch (RepositoryException re ) {
// something went wrong, so let's log it
this.logger.error("Exception during writing new job '" + nodeName + "' to repository.", re);
}
}
// if we were able to write the event into the repository
// we will queue it for processing
if ( info.nodePath != null ) {
try {
this.queue.put(info);
} catch (InterruptedException e) {
// this should never happen
this.ignoreException(e);
}
}
}
}
}
/**
* This method runs in the background and processes the local queue.
*/
protected void runInBackground() throws RepositoryException {
this.backgroundSession = this.createSession();
this.backgroundSession.getWorkspace().getObservationManager()
.addEventListener(this,
javax.jcr.observation.Event.PROPERTY_REMOVED,
this.repositoryPath,
true,
null,
new String[] {this.getEventNodeType()},
true);
// load unprocessed jobs from repository
this.loadJobs();
while ( this.running ) {
// so let's wait/get the next job 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 ) {
// check if the node still exists
synchronized (this.backgroundSession) {
try {
this.backgroundSession.refresh(false);
if ( this.backgroundSession.itemExists(info.nodePath) ) {
final Event event = info.event;
final String jobTopic = (String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
final boolean parallelProcessing = event.getProperty(EventUtil.PROPERTY_JOB_PARALLEL) != null;
// check how we can process this job
// if parallel processing is allowed, we can just process
// if not we should check if any other job with the same topic is currently running
boolean process = parallelProcessing;
if ( !process ) {
synchronized ( this.processingMap ) {
final Boolean value = this.processingMap.get(jobTopic);
if ( value == null || !value.booleanValue() ) {
this.processingMap.put(jobTopic, Boolean.TRUE);
process = true;
}
}
}
if ( process ) {
boolean unlock = true;
try {
final Node eventNode = (Node) this.backgroundSession.getItem(info.nodePath);
if ( !eventNode.isLocked() ) {
// lock node
try {
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
process = false;
}
if ( process ) {
unlock = false;
this.processJob(info.event, eventNode);
}
}
} catch (RepositoryException e) {
// ignore
this.ignoreException(e);
} finally {
if ( unlock && !parallelProcessing ) {
synchronized ( this.processingMap ) {
this.processingMap.put(jobTopic, Boolean.FALSE);
}
}
}
} else {
try {
// check if the node is in processing or already finished
final Node eventNode = (Node) this.backgroundSession.getItem(info.nodePath);
if ( !eventNode.isLocked() ) {
try {
this.queue.put(info);
} catch (InterruptedException e) {
// ignore
this.ignoreException(e);
}
// wait time before we restart the cycle, if there is only one job in the queue!
if ( this.queue.size() == 1 ) {
try {
Thread.sleep(this.sleepTime);
} catch (InterruptedException e) {
// ignore
this.ignoreException(e);
}
}
}
} catch (RepositoryException e) {
// ignore
this.ignoreException(e);
}
}
}
} catch (RepositoryException re) {
this.ignoreException(re);
}
}
}
}
}
/**
* @see org.apache.sling.engine.event.impl.JobPersistenceHandler#getContainerNodeType()
*/
protected String getContainerNodeType() {
return EventHelper.JOBS_NODE_TYPE;
}
/**
* @see org.apache.sling.engine.event.impl.JobPersistenceHandler#getEventNodeType()
*/
protected String getEventNodeType() {
return EventHelper.JOB_NODE_TYPE;
}
/**
* @see org.osgi.service.event.EventHandler#handleEvent(org.osgi.service.event.Event)
*/
public void handleEvent(final Event event) {
// we ignore remote job events
if ( EventUtil.isLocal(event) ) {
// check for bundle event
if ( event.getTopic().equals(EventUtil.TOPIC_JOB)) {
// job event
final String jobTopic = (String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
// job topic must be set, otherwise we ignore this event!
if ( jobTopic != null ) {
// queue the event in order to respond quickly
try {
this.writeQueue.put(event);
} catch (InterruptedException e) {
// this should never happen
this.ignoreException(e);
}
} else {
this.logger.warn("Event does not contain job topic: {}", event);
}
} else {
// bundle event started or updated
boolean doIt = false;
synchronized ( this.unloadedJobs ) {
if ( this.unloadedJobs.size() > 0 ) {
doIt = true;
}
}
if ( doIt ) {
final Runnable t = new Runnable() {
public void run() {
synchronized (unloadedJobs) {
Session s = null;
final Set<String> newUnloadedJobs = new HashSet<String>();
newUnloadedJobs.addAll(unloadedJobs);
try {
s = createSession();
for(String path : unloadedJobs ) {
newUnloadedJobs.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) {
newUnloadedJobs.add(path);
ignoreException(cnfe);
}
}
}
} catch (RepositoryException re) {
// we ignore this and readd
newUnloadedJobs.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();
}
unloadedJobs.clear();
unloadedJobs.addAll(newUnloadedJobs);
}
}
}
};
this.threadPool.execute(t);
}
}
}
}
public static final String ALLOWED_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ abcdefghijklmnopqrstuvwxyz0123456789_,.-+*#!�$%&()=[]?";
public static final char REPLACEMENT_CHAR = '_';
public static String filter(final String nodeName) {
final StringBuffer sb = new StringBuffer();
char lastAdded = 0;
for(int i=0; i < nodeName.length(); i++) {
final char c = nodeName.charAt(i);
char toAdd = c;
if (ALLOWED_CHARS.indexOf(c) < 0) {
if (lastAdded == REPLACEMENT_CHAR) {
// do not add several _ in a row
continue;
}
toAdd = REPLACEMENT_CHAR;
} else if(i == 0 && Character.isDigit(c)) {
sb.append(REPLACEMENT_CHAR);
}
sb.append(toAdd);
lastAdded = toAdd;
}
if (sb.length()==0) {
sb.append(REPLACEMENT_CHAR);
}
return sb.toString();
}
/**
* Create a unique node name for the job.
*/
protected String getNodeName(Event event) {
final String jobId = (String)event.getProperty(EventUtil.PROPERTY_JOB_ID);
final String name;
if ( jobId != null ) {
final String jobTopic = ((String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC));
name = jobTopic + " " + jobId;
} else {
name = "Job " + UUID.randomUUID().toString();
}
return filter(name);
}
/**
* Process a job and unlock the node in the repository.
* @param event The original event.
* @param eventNode The node in the repository where the job is stored.
*/
protected void processJob(Event event, Node eventNode) {
final boolean parallelProcessing = event.getProperty(EventUtil.PROPERTY_JOB_PARALLEL) != null;
final String jobTopic = (String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
boolean unlock = true;
try {
final Event jobEvent = this.getJobEvent(event, eventNode.getPath());
eventNode.setProperty(EventHelper.NODE_PROPERTY_PROCESSOR, this.applicationId);
eventNode.save();
final EventAdmin localEA = this.eventAdmin;
if ( localEA != null ) {
localEA.sendEvent(jobEvent);
// do not unlock if sending was successful
unlock = false;
} else {
this.logger.error("Job event can't be sent as no event admin is available.");
}
} catch (RepositoryException re) {
// if an exception occurs, we just log
this.logger.error("Exception during job processing.", re);
} finally {
if ( unlock ) {
if ( !parallelProcessing ) {
synchronized ( this.processingMap ) {
this.processingMap.put(jobTopic, Boolean.FALSE);
}
}
// unlock node
try {
eventNode.unlock();
} catch (RepositoryException e) {
// if unlock fails, we silently ignore this
this.ignoreException(e);
}
}
}
}
/**
* Create the job event.
* @param e
* @return
*/
protected Event getJobEvent(Event e, String nodePath) {
final String eventTopic = (String)e.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
final Dictionary<String, Object> properties = new Hashtable<String, Object>();
final String[] propertyNames = e.getPropertyNames();
for(int i=0; i<propertyNames.length; i++) {
properties.put(propertyNames[i], e.getProperty(propertyNames[i]));
}
// put properties for finished job callback
properties.put(EventUtil.JobStatusNotifier.CONTEXT_PROPERTY_NAME, new EventUtil.JobStatusNotifier.NotifierContext(this, nodePath));
return new Event(eventTopic, properties);
}
/**
* @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_JOB_TOPIC));
final String jobId = (String)event.getProperty(EventUtil.PROPERTY_JOB_ID);
if ( jobId != null ) {
eventNode.setProperty(EventHelper.NODE_PROPERTY_JOBID, jobId);
}
}
/**
* @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) {
try {
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) ) {
boolean doNotProcess = false;
synchronized ( this.deletedJobs ) {
doNotProcess = this.deletedJobs.remove(nodePath);
}
if ( !doNotProcess ) {
final Node eventNode = (Node) s.getItem(nodePath);
if ( !eventNode.isLocked() && !eventNode.hasProperty(EventHelper.NODE_PROPERTY_FINISHED)) {
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) {
// store path for lazy loading
synchronized ( this.unloadedJobs ) {
this.unloadedJobs.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();
}
}
}
/**
* Load all active jobs from the repository.
* @throws RepositoryException
*/
protected void loadJobs() {
try {
final QueryManager qManager = this.backgroundSession.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() && !eventNode.hasProperty(EventHelper.NODE_PROPERTY_FINISHED)) {
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) {
// store path for lazy loading
synchronized ( this.unloadedJobs ) {
this.unloadedJobs.add(nodePath);
}
this.ignoreException(cnfe);
} catch (RepositoryException re) {
this.logger.error("Unable to load stored job from " + nodePath, re);
}
}
}
} catch (RepositoryException re) {
this.logger.error("Exception during initial loading of stored jobs.", re);
}
}
/**
* @see org.apache.sling.event.EventUtil.JobStatusNotifier#finishedJob(org.osgi.service.event.Event, String, boolean)
*/
public boolean finishedJob(Event job, String eventNodePath, boolean shouldReschedule) {
boolean reschedule = shouldReschedule;
if ( shouldReschedule ) {
// check if we exceeded the number of retries
int retries = this.maxJobRetries;
if ( job.getProperty(EventUtil.PROPERTY_JOB_RETRIES) != null ) {
retries = (Integer) job.getProperty(EventUtil.PROPERTY_JOB_RETRIES);
}
int retryCount = 0;
if ( job.getProperty(EventUtil.PROPERTY_JOB_RETRY_COUNT) != null ) {
retryCount = (Integer)job.getProperty(EventUtil.PROPERTY_JOB_RETRY_COUNT);
}
retryCount++;
if ( retryCount > retries ) {
reschedule = false;
}
// update event with retry count
final Dictionary<String, Object> newProperties;
// create a new dictionary
newProperties = new Hashtable<String, Object>();
final String[] names = job.getPropertyNames();
for(int i=0; i<names.length; i++ ) {
newProperties.put(names[i], job.getProperty(names[i]));
}
newProperties.put(EventUtil.PROPERTY_JOB_RETRY_COUNT, retryCount);
job = new Event(job.getTopic(), newProperties);
}
final boolean parallelProcessing = job.getProperty(EventUtil.PROPERTY_JOB_PARALLEL) != null;
// we have to use the same session for unlocking that we used for locking!
synchronized ( this.backgroundSession ) {
try {
this.backgroundSession.refresh(false);
final Node eventNode = (Node) this.backgroundSession.getItem(eventNodePath);
boolean unlock = true;
try {
if ( !reschedule ) {
synchronized ( this.deletedJobs ) {
this.deletedJobs.add(eventNodePath);
}
// unlock node
try {
eventNode.unlock();
} catch (RepositoryException e) {
// if unlock fails, we silently ignore this
this.ignoreException(e);
}
unlock = false;
final String jobId = (String)job.getProperty(EventUtil.PROPERTY_JOB_ID);
if ( jobId == null ) {
// remove node from repository if no job id set
final Node parentNode = eventNode.getParent();
eventNode.remove();
parentNode.save();
} else {
eventNode.setProperty(EventHelper.NODE_PROPERTY_FINISHED, Calendar.getInstance());
eventNode.save();
}
}
} catch (RepositoryException re) {
// if an exception occurs, we just log
this.logger.error("Exception during job finishing.", re);
} finally {
if ( !parallelProcessing) {
final String jobTopic = (String)job.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
synchronized ( this.processingMap ) {
this.processingMap.put(jobTopic, Boolean.FALSE);
}
}
if ( unlock ) {
synchronized ( this.deletedJobs ) {
this.deletedJobs.add(eventNodePath);
}
// unlock node
try {
eventNode.unlock();
} catch (RepositoryException e) {
// if unlock fails, we silently ignore this
this.ignoreException(e);
}
}
}
if ( reschedule ) {
final EventInfo info = new EventInfo();
try {
info.event = job;
info.nodePath = eventNode.getPath();
} catch (RepositoryException e) {
// this should never happen
this.ignoreException(e);
}
// delay rescheduling?
if ( job.getProperty(EventUtil.PROPERTY_JOB_RETRY_DELAY) != null ) {
final long delay = (Long)job.getProperty(EventUtil.PROPERTY_JOB_RETRY_DELAY);
final Runnable t = new Runnable() {
public void run() {
try {
Thread.sleep(delay);
} catch (InterruptedException e) {
// this should never happen
ignoreException(e);
}
try {
queue.put(info);
} catch (InterruptedException e) {
// this should never happen
ignoreException(e);
}
}
};
this.threadPool.execute(t);
} else {
// put directly into queue
try {
this.queue.put(info);
} catch (InterruptedException e) {
// this should never happen
this.ignoreException(e);
}
}
}
if ( !shouldReschedule ) {
return true;
}
return reschedule;
} catch (RepositoryException re) {
this.logger.error("Unable to create new session.", re);
return false;
}
}
}
/**
* @see org.apache.sling.event.EventUtil.JobStatusNotifier#execute(java.lang.Runnable)
*/
public void execute(Runnable job) {
this.threadPool.execute(job);
}
/**
* Search for active nodes
* @param topic
* @return
* @throws RepositoryException
*/
protected Collection<Event> queryCurrentJobs(String topic, boolean locked) {
// 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);
buffer.append("//element(*, ");
buffer.append(this.getEventNodeType());
buffer.append(")");
if ( topic != null ) {
buffer.append(" [");
buffer.append(EventHelper.NODE_PROPERTY_TOPIC);
buffer.append(" = '");
buffer.append(topic);
buffer.append("'");
}
if ( locked ) {
buffer.append(" and ");
buffer.append("jcr:lockOwner");
}
buffer.append("]");
final Query q = qManager.createQuery(buffer.toString(), 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.JobStatusProvider#getCurrentJobs(java.lang.String)
*/
public Collection<Event> getCurrentJobs(String topic) {
return this.queryCurrentJobs(topic, true);
}
/**
* @see org.apache.sling.event.JobStatusProvider#scheduledJobs(java.lang.String)
*/
public Collection<Event> scheduledJobs(String topic) {
return this.queryCurrentJobs(topic, false);
}
}