/**
* @see org.apache.sling.event.impl.AbstractRepositoryEventHandler#runInBackground()
*/
protected void runInBackground() {
Event event = null;
while ( this.running ) {
// so let's wait/get the next event from the queue
if ( event == null ) {
try {
event = this.queue.take();
} catch (final InterruptedException e) {
this.ignoreException(e);
Thread.currentThread().interrupt();
this.running = false;
}
}
if ( event != null && this.running ) {
// check event type
if ( event.getTopic().equals(SlingConstants.TOPIC_RESOURCE_ADDED)
|| event.getTopic().equals(SlingConstants.TOPIC_RESOURCE_CHANGED)) {
final String path = (String)event.getProperty(SlingConstants.PROPERTY_PATH);
event = null;
ResourceResolver resolver = null;
try {
resolver = this.resourceResolverFactory.getAdministrativeResourceResolver(null);
final Resource eventResource = resolver.getResource(path);
if ( TimedEventReceiver.TIMED_EVENT_RESOURCE_TYPE.equals(eventResource.getResourceType()) ) {
final ReadResult result = this.readEvent(eventResource);
if ( result != null ) {
if ( result.hasReadErrors ) {
synchronized ( this.unloadedEvents ) {
this.unloadedEvents.add(eventResource.getPath());
}
} else {
event = result.event;
}
}
}
} catch (final LoginException le) {
this.ignoreException(le);
} finally {
if ( resolver != null ) {
resolver.close();
}
}
} else if ( event.getTopic().equals(SlingConstants.TOPIC_RESOURCE_REMOVED) ) {
final String path = (String)event.getProperty(SlingConstants.PROPERTY_PATH);
final String jobId = ResourceUtil.getName(path);
this.startedSchedulerJobs.remove(jobId);
logger.debug("Stopping job with id : {}", jobId);
this.scheduler.unschedule(jobId);
event = null;
} else if ( !NotificationConstants.TOPIC_JOB_FINISHED.equals(event.getTopic()) ) {
ScheduleInfo scheduleInfo = null;
try {
scheduleInfo = new ScheduleInfo(event);
} catch (final IllegalArgumentException iae) {
this.logger.error(iae.getMessage());
}
if ( scheduleInfo != null ) {
// if something went wrong, we reschedule
if ( !this.processEvent(event, scheduleInfo) ) {
try {
this.queue.put(event);
} catch (final InterruptedException e) {
this.ignoreException(e);
Thread.currentThread().interrupt();
this.running = false;
}
}
}
event = null;
} else if (NotificationConstants.TOPIC_JOB_FINISHED.equals(event.getTopic())){
// stopScheduling() puts this event on the queue, but the intention is unclear to me.
// as the threadStarted flag ensures the background thread is only started once, we must not stop
// the thread, otherwise its never started again upon topology changes.
event = null;
} else {