Author: cziegeler Date: Mon Sep 30 13:40:46 2013 New Revision: 1527561 URL: http://svn.apache.org/r1527561 Log: SLING-3028 : Support for progress tracking of jobs
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobImpl.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/AbstractJobQueue.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/Job.java sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/JobManager.java Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java Mon Sep 30 13:40:46 2013 @@ -72,6 +72,12 @@ public class JobHandler { this.jobManager.reassign(this.job); } + public void persistJobProperties(final String... propNames) { + if ( propNames != null ) { + this.jobManager.persistJobProperties(this.job, propNames); + } + } + @Override public int hashCode() { return this.job.getId().hashCode(); @@ -89,9 +95,4 @@ public class JobHandler { public String toString() { return "JobHandler(" + this.job.getId() + ")"; } - - public void updateProperty(final String propName) { - this.jobManager.updateProperty(this.job, propName); - - } } \ No newline at end of file Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobImpl.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobImpl.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobImpl.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobImpl.java Mon Sep 30 13:40:46 2013 @@ -46,27 +46,6 @@ public class JobImpl implements Job { /** Internal job property containing optional delay override. */ public static final String PROPERTY_DELAY_OVERRIDE = ":slingevent:delayOverride"; - /** Property for log statements. */ - public static final String PROPERTY_LOG = "slingevent:log"; - - /** Property for ETA. */ - public static final String PROPERTY_ETA = "slingevent:eta"; - - /** Property for Steps. */ - public static final String PROPERTY_STEPS = "slingevent:steps"; - - /** Property for Step. */ - public static final String PROPERTY_STEP = "slingevent:step"; - - /** Property for final message. */ - public static final String PROPERTY_MESSAGE = "slingevent:message"; - - /** Property for finished jobs. */ - public static final String PROPERTY_FINISHED_STATE = "slingevent:finishedState"; - - /** Property for finished jobs. */ - public static final String PROPERTY_FINISHED_DATE = "slingevent:finishedDate"; - private final ValueMap properties; private final String topic; @@ -260,45 +239,56 @@ public class JobImpl implements Job { */ public void prepare() { this.properties.remove(JobImpl.PROPERTY_DELAY_OVERRIDE); - this.properties.remove(JobImpl.PROPERTY_LOG); - this.properties.remove(JobImpl.PROPERTY_ETA); - this.properties.remove(JobImpl.PROPERTY_STEPS); - this.properties.remove(JobImpl.PROPERTY_STEP); - this.properties.remove(JobImpl.PROPERTY_MESSAGE); + this.properties.remove(Job.PROPERTY_JOB_LOG); + this.properties.remove(Job.PROPERTY_JOB_PROGRESS_ETA); + this.properties.remove(Job.PROPERTY_JOB_PROGRESS_STEPS); + this.properties.remove(Job.PROPERTY_JOB_PROGRESS_STEP); + this.properties.remove(Job.PROPERTY_RESULT_MESSAGE); } - public String update(final long eta) { - this.setProperty(JobImpl.PROPERTY_ETA, eta); - return JobImpl.PROPERTY_ETA; + public String[] startProgress(final int steps, final long eta) { + if ( steps > 0 ) { + this.setProperty(Job.PROPERTY_JOB_PROGRESS_STEPS, steps); + } + if ( eta > 0 ) { + this.setProperty(Job.PROPERTY_JOB_PROGRESS_ETA, eta); + } + return new String[] {Job.PROPERTY_JOB_PROGRESS_ETA, PROPERTY_JOB_PROGRESS_STEPS}; } - public String startProgress(final long eta) { - this.setProperty(JobImpl.PROPERTY_ETA, eta); - return JobImpl.PROPERTY_ETA; - } + public String setProgress(final int step) { + final int steps = this.getProperty(Job.PROPERTY_JOB_PROGRESS_STEPS, -1); + if ( steps > 0 && step > 0 ) { + int current = this.getProperty(Job.PROPERTY_JOB_PROGRESS_STEP, 0); + current += step; + if ( current > steps ) { + current = steps; + } + this.setProperty(Job.PROPERTY_JOB_PROGRESS_STEP, current); - public String startProgress(final int steps) { - this.setProperty(JobImpl.PROPERTY_STEPS, steps); - return JobImpl.PROPERTY_STEPS; + // TODO - recalculate ETA + return Job.PROPERTY_JOB_PROGRESS_STEP; + } + return null; } - public String setProgress(final int step) { - this.setProperty(JobImpl.PROPERTY_STEP, step); - return JobImpl.PROPERTY_STEP; + public String update(final long eta) { + this.setProperty(Job.PROPERTY_JOB_PROGRESS_ETA, eta); + return Job.PROPERTY_JOB_PROGRESS_ETA; } - public String log(final String message, Object... args) { + public String log(final String message, final Object... args) { final String logEntry = MessageFormat.format(message, args); - final String[] entries = this.getProperty(JobImpl.PROPERTY_LOG, String[].class); + final String[] entries = this.getProperty(Job.PROPERTY_JOB_LOG, String[].class); if ( entries == null ) { - this.setProperty(JobImpl.PROPERTY_LOG, new String[] {logEntry}); + this.setProperty(Job.PROPERTY_JOB_LOG, new String[] {logEntry}); } else { final String[] newEntries = new String[entries.length + 1]; System.arraycopy(entries, 0, newEntries, 0, entries.length); newEntries[entries.length] = logEntry; - this.setProperty(JobImpl.PROPERTY_LOG, newEntries); + this.setProperty(Job.PROPERTY_JOB_LOG, newEntries); } - return JobImpl.PROPERTY_LOG; + return Job.PROPERTY_JOB_LOG; } @Override Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java Mon Sep 30 13:40:46 2013 @@ -506,7 +506,12 @@ public class JobManagerImpl jobProperties.put(Job.PROPERTY_JOB_RETRIES, vm.get(Job.PROPERTY_JOB_RETRIES, Integer.class)); jobProperties.put(Job.PROPERTY_JOB_RETRY_COUNT, vm.get(Job.PROPERTY_JOB_RETRY_COUNT, Integer.class)); jobProperties.put(Job.PROPERTY_JOB_PRIORITY, JobPriority.valueOf(vm.get(Job.PROPERTY_JOB_PRIORITY, JobPriority.NORM.name()))); - + if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS) != null ) { + jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEPS, vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS, Integer.class)); + } + if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEP) != null ) { + jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEP, vm.get(Job.PROPERTY_JOB_PROGRESS_STEP, Integer.class)); + } @SuppressWarnings("unchecked") final List<Exception> readErrorList = (List<Exception>) jobProperties.get(ResourceHelper.PROPERTY_MARKER_READ_ERROR_LIST); if ( readErrorList != null ) { @@ -1068,7 +1073,9 @@ public class JobManagerImpl final Map<String, Object> props = new HashMap<String, Object>(vm); props.put(JobImpl.PROPERTY_FINISHED_STATE, isSuccess ? JobState.SUCCEEDED.name() : JobState.CANCELLED.name()); props.put(JobImpl.PROPERTY_FINISHED_DATE, Calendar.getInstance()); - + if ( job.getProperty(Job.PROPERTY_RESULT_MESSAGE) != null ) { + props.put(Job.PROPERTY_RESULT_MESSAGE, job.getProperty(Job.PROPERTY_RESULT_MESSAGE)); + } ResourceHelper.getOrCreateResource(resolver, newPath, props); } resolver.delete(jobResource); @@ -1112,6 +1119,9 @@ public class JobManagerImpl if ( jobResource != null ) { final ModifiableValueMap mvm = jobResource.adaptTo(ModifiableValueMap.class); mvm.put(Job.PROPERTY_JOB_RETRY_COUNT, job.getProperty(Job.PROPERTY_JOB_RETRY_COUNT)); + if ( job.getProperty(Job.PROPERTY_RESULT_MESSAGE) != null ) { + mvm.put(Job.PROPERTY_RESULT_MESSAGE, job.getProperty(Job.PROPERTY_RESULT_MESSAGE)); + } mvm.remove(Job.PROPERTY_JOB_STARTED_TIME); try { resolver.commit(); @@ -1145,11 +1155,11 @@ public class JobManagerImpl mvm.put(Job.PROPERTY_JOB_QUEUE_NAME, info.getJob().getQueueName()); mvm.put(Job.PROPERTY_JOB_RETRIES, info.getJob().getNumberOfRetries()); mvm.put(Job.PROPERTY_JOB_PRIORITY, info.getJob().getJobPriority().name()); - mvm.remove(JobImpl.PROPERTY_ETA); - mvm.remove(JobImpl.PROPERTY_STEPS); - mvm.remove(JobImpl.PROPERTY_STEP); - mvm.remove(JobImpl.PROPERTY_LOG); - mvm.remove(JobImpl.PROPERTY_MESSAGE); + mvm.remove(Job.PROPERTY_JOB_PROGRESS_ETA); + mvm.remove(Job.PROPERTY_JOB_PROGRESS_STEPS); + mvm.remove(Job.PROPERTY_JOB_PROGRESS_STEP); + mvm.remove(Job.PROPERTY_JOB_LOG); + mvm.remove(Job.PROPERTY_RESULT_MESSAGE); resolver.commit(); return true; @@ -1387,14 +1397,19 @@ public class JobManagerImpl /** * Update the property of a job in the resource tree */ - public void updateProperty(final JobImpl job, final String propName) { + public void persistJobProperties(final JobImpl job, final String... propNames) { ResourceResolver resolver = null; try { resolver = this.resourceResolverFactory.getAdministrativeResourceResolver(null); final Resource jobResource = resolver.getResource(job.getResourcePath()); if ( jobResource != null ) { final ModifiableValueMap mvm = jobResource.adaptTo(ModifiableValueMap.class); - mvm.put(propName, job.getProperty(propName)); + for(final String propName : propNames) { + final Object val = job.getProperty(propName); + if ( val != null ) { + mvm.put(propName, job.getProperty(propName)); + } + } resolver.commit(); } } catch ( final PersistenceException ignore ) { @@ -1413,7 +1428,7 @@ public class JobManagerImpl */ @Override public void stopJobById(final String jobId) { - // not implemented yet - + // TODO not implemented yet + throw new IllegalStateException("Not implemented yet..."); } } Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/AbstractJobQueue.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/AbstractJobQueue.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/AbstractJobQueue.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/AbstractJobQueue.java Mon Sep 30 13:40:46 2013 @@ -543,26 +543,34 @@ public abstract class AbstractJobQueue synchronized ( lock ) { result = consumer.process(job, new JobExecutionContext() { + private boolean hasInit = false; + @Override public void initProgress(final int steps, final long eta) { - handler.updateProperty(job.startProgress(steps)); - handler.updateProperty(job.startProgress(eta)); + if ( !hasInit ) { + handler.persistJobProperties(job.startProgress(steps, eta)); + hasInit = true; + } } @Override public void incrementProgressCount(final int steps) { - handler.updateProperty(job.setProgress(steps)); + if ( hasInit ) { + handler.persistJobProperties(job.setProgress(steps)); + } } @Override public void updateProgress(final long eta) { - handler.updateProperty(job.update(eta)); + if ( hasInit ) { + handler.persistJobProperties(job.update(eta)); + } } @Override public void log(final String message, Object... args) { - handler.updateProperty(job.log(message, args)); + handler.persistJobProperties(job.log(message, args)); } @Override @@ -599,6 +607,9 @@ public abstract class AbstractJobQueue if ( result.getRetryDelayInMs() != null ) { job.setProperty(JobImpl.PROPERTY_DELAY_OVERRIDE, result.getRetryDelayInMs()); } + if ( result.getMessage() != null ) { + job.setProperty(Job.PROPERTY_RESULT_MESSAGE, result.getMessage()); + } finishedJob(job.getId(), result, false); } } Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java Mon Sep 30 13:40:46 2013 @@ -36,6 +36,7 @@ import org.apache.sling.api.resource.Val import org.apache.sling.event.EventUtil; import org.apache.sling.event.impl.jobs.JobImpl; import org.apache.sling.event.impl.jobs.deprecated.JobStatusNotifier; +import org.apache.sling.event.jobs.Job; import org.apache.sling.event.jobs.JobUtil; import org.apache.sling.event.jobs.consumer.JobConsumer; import org.osgi.service.event.EventConstants; @@ -65,7 +66,14 @@ public abstract class ResourceHelper { JobUtil.PROPERTY_NOTIFICATION_JOB, JobStatusNotifier.CONTEXT_PROPERTY_NAME, JobImpl.PROPERTY_DELAY_OVERRIDE, - JobConsumer.PROPERTY_JOB_ASYNC_HANDLER + JobConsumer.PROPERTY_JOB_ASYNC_HANDLER, + Job.PROPERTY_JOB_LOG, + Job.PROPERTY_JOB_PROGRESS_ETA, + Job.PROPERTY_JOB_PROGRESS_STEP, + Job.PROPERTY_JOB_PROGRESS_STEPS, + Job.PROPERTY_FINISHED_DATE, + Job.PROPERTY_FINISHED_STATE, + Job.PROPERTY_RESULT_MESSAGE }; /** Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/Job.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/Job.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/Job.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/Job.java Mon Sep 30 13:40:46 2013 @@ -1,5 +1,5 @@ /* - * Licensed to the Apache Software Foundation (ASF) under one + 1 * 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 @@ -116,6 +116,58 @@ public interface Job { String PROPERTY_JOB_RETRY_DELAY = "event.job.retrydelay"; /** + * This property contains the optional output log of a job consumer. + * The value of this property is a string array. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_JOB_LOG = "slingevent:log"; + + /** + * This property contains the optional ETA for a job. + * The value of this property is a {@link Calendar} object. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_JOB_PROGRESS_ETA = "slingevent:progressETA"; + + /** + * This property contains optional progress information about a job, + * the number of steps the job consumer will perform. Each step is + * assumed to consume roughly the same amount if time. + * The value of this property is an integer. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_JOB_PROGRESS_STEPS = "slingevent:progressSteps"; + + /** + * This property contains optional progress information about a job, + * the number of completed steps. + * The value of this property is an integer. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_JOB_PROGRESS_STEP = "slingevent:progressStep"; + + /** + * This property contains the optional result message of a job consumer. + * The value of this property is a string. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_RESULT_MESSAGE = "slingevent:resultMessage"; + + /** + * This property contains the finished state of a job once it's marked as finished. + * TODO - DOCUMENT + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_FINISHED_STATE = "slingevent:finishedState"; + + /** + * This property contains the finished date once a job is marked as finished. + * The value of this property is a {@link Calendar} object. + * This property is read-only and can't be specified when the job is created. + */ + String PROPERTY_FINISHED_DATE = "slingevent:finishedDate"; + + /** * The job topic. * @return The job topic */ Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/JobManager.java URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/JobManager.java?rev=1527561&r1=1527560&r2=1527561&view=diff ============================================================================== --- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/JobManager.java (original) +++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/jobs/JobManager.java Mon Sep 30 13:40:46 2013 @@ -85,6 +85,9 @@ public interface JobManager { * A job topic is a hierarchical name separated by dashes, each part has to start with a letter, * allowed characters are letters, numbers and the underscore. * + * The returned job object is a snapshot of the job state taken at the time of creation. Updates + * to the job state are not reflected and the client needs to get a new job object using the job id. + * * @param topic The required job topic. * @param properties Optional job properties. The properties must be serializable. * @return The new job - or <code>null</code> if the job could not be created. @@ -107,6 +110,9 @@ public interface JobManager { * dealing with this situation and as jobs with name come with a heavy processing overhead * these should be avoided. * + * The returned job object is a snapshot of the job state taken at the time of creation. Updates + * to the job state are not reflected and the client needs to get a new job object using the job id. + * * @param topic The required job topic. * @param name Optional unique job name * @param properties Optional job properties. The properties must be serializable. @@ -116,12 +122,22 @@ public interface JobManager { Job addJob(String topic, String name, Map<String, Object> properties); /** + * Return a job based on the unique job name. + * + * The returned job object is a snapshot of the job state taken at the time of the call. Updates + * to the job state are not reflected and the client needs to get a new job object using the job id. + * * @return A job or <code>null</code> * @since 1.2 */ Job getJobByName(String name); /** + * Return a job based on the unique id. + * + * The returned job object is a snapshot of the job state taken at the time of the call. Updates + * to the job state are not reflected and the client needs to get a new job object using the job id. + * * @param jobId The unique identifier from {@link Job#getId()} * @return A job or <code>null</code> * @since 1.2 @@ -130,9 +146,11 @@ public interface JobManager { /** * Removes the job even if it is currently in processing. + * * If the job exists and is not in processing, it gets removed from the processing queue. * If the job exists and is in processing, it is removed from the persistence layer, * however processing is not stopped. + * * @param jobId The unique identifier from {@link Job#getId()} * @return <code>true</code> if the job could be removed or does not exist anymore. * <code>false</code> otherwise. @@ -142,9 +160,13 @@ public interface JobManager { /** * Find a job - either scheduled or active. + * * This method searches for an event with the given topic and filter properties. If more than one * job matches, the first one found is returned which could be any of the matching jobs. * + * The returned job object is a snapshot of the job state taken at the time of the call. Updates + * to the job state are not reflected and the client needs to get a new job object using the job id. + * * @param topic Topic is required. * @param template The map acts like a template. The searched job * must match the template (AND query). @@ -163,6 +185,9 @@ public interface JobManager { * If the history is returned, the result set is sorted in descending order, listening the newest entry * first. For unfinished jobs, the result set is sorted in ascending order. * + * The returned job objects are a snapshot of the jobs state taken at the time of the call. Updates + * to the job states are not reflected and the client needs to get new job objects. + * * @param type Required parameter for the type. See above. * @param topic Topic can be used as a filter, if it is non-null, only jobs with this topic will be returned. * @param limit A positive number indicating the maximum number of jobs returned by the iterator. A value