You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by ro...@apache.org on 2017/11/07 09:50:38 UTC
[sling-org-apache-sling-jobs] annotated tag
org.apache.sling.jobs-1.0.0 created (now a8cf43e)
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a change to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git.
at a8cf43e (tag)
tagging 0f219926b0329fa6762cf6575e2f36e0835a8eb1 (commit)
by Ian Boston
on Wed Oct 5 10:05:53 2016 +0000
- Log -----------------------------------------------------------------
org.apache.sling.jobs-1.0.0
-----------------------------------------------------------------------
This annotated tag includes the following new commits:
new 76cbcf0 SLING-5645 moved jobs out of examples
new 8e498cc SLING-5645 added provider types to apis where applicable
new 09609b0 SLING-5645 Improved IT tests, improved config checking and reduced dependency footprint
new d4a0225 SLING-5645 fixes to javadoc
new f74a7ad [maven-release-plugin] prepare release org.apache.sling.jobs-1.0.0
new 0f21992 [maven-release-plugin] copy for tag org.apache.sling.jobs-1.0.0
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
--
To stop receiving notification emails like this one, please contact
['"commits@sling.apache.org" <co...@sling.apache.org>'].
[sling-org-apache-sling-jobs] 06/06: [maven-release-plugin] copy
for tag org.apache.sling.jobs-1.0.0
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit 0f219926b0329fa6762cf6575e2f36e0835a8eb1
Author: Ian Boston <ie...@apache.org>
AuthorDate: Wed Oct 5 10:05:53 2016 +0000
[maven-release-plugin] copy for tag org.apache.sling.jobs-1.0.0
git-svn-id: https://svn.apache.org/repos/asf/sling/tags/org.apache.sling.jobs-1.0.0@1763387 13f79535-47bb-0310-9956-ffa450edef68
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.
[sling-org-apache-sling-jobs] 01/06: SLING-5645 moved jobs out of
examples
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit 76cbcf0f872b745b08a3cb6bd34aee5b3e08a138
Author: Ian Boston <ie...@apache.org>
AuthorDate: Mon Oct 3 16:05:02 2016 +0000
SLING-5645 moved jobs out of examples
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core@1763181 13f79535-47bb-0310-9956-ffa450edef68
---
README.md | 27 ++
pom.xml | 142 ++++++++
src/main/java/org/apache/sling/jobs/Job.java | 160 +++++++++
.../java/org/apache/sling/jobs/JobBuilder.java | 44 +++
.../java/org/apache/sling/jobs/JobCallback.java | 33 ++
.../java/org/apache/sling/jobs/JobConsumer.java | 46 +++
.../java/org/apache/sling/jobs/JobController.java | 43 +++
.../java/org/apache/sling/jobs/JobManager.java | 165 +++++++++
.../java/org/apache/sling/jobs/JobTypeValve.java | 36 ++
src/main/java/org/apache/sling/jobs/JobUpdate.java | 162 +++++++++
.../org/apache/sling/jobs/JobUpdateBuilder.java | 48 +++
.../org/apache/sling/jobs/JobUpdateListener.java | 34 ++
src/main/java/org/apache/sling/jobs/Types.java | 183 ++++++++++
.../org/apache/sling/jobs/impl/JobBuilderImpl.java | 68 ++++
.../java/org/apache/sling/jobs/impl/JobImpl.java | 244 +++++++++++++
.../org/apache/sling/jobs/impl/JobManagerImpl.java | 123 +++++++
.../sling/jobs/impl/JobQueueConsumerFactory.java | 114 ++++++
.../org/apache/sling/jobs/impl/JobSubsystem.java | 199 +++++++++++
.../sling/jobs/impl/JobUpdateBuilderImpl.java | 107 ++++++
.../org/apache/sling/jobs/impl/JobUpdateImpl.java | 234 +++++++++++++
.../apache/sling/jobs/impl/ManagerSubscriber.java | 51 +++
.../sling/jobs/impl/OutboundJobUpdateListener.java | 65 ++++
.../java/org/apache/sling/jobs/impl/Utils.java | 150 ++++++++
.../org/apache/sling/jobs/impl/spi/JobStarter.java | 28 ++
.../org/apache/sling/jobs/impl/spi/JobStorage.java | 70 ++++
.../sling/jobs/impl/spi/MapValueAdapter.java | 45 +++
.../jobs/impl/storage/InMemoryJobStorage.java | 79 +++++
.../java/org/apache/sling/jobs/package-info.java | 23 ++
.../apache/sling/jobs/impl/JobBuilderImplTest.java | 97 ++++++
.../org/apache/sling/jobs/impl/JobImplTest.java | 387 +++++++++++++++++++++
.../apache/sling/jobs/impl/JobManagerImplTest.java | 264 ++++++++++++++
.../apache/sling/jobs/impl/JobUpdateImplTest.java | 81 +++++
.../sling/jobs/impl/ManagerSubscriberTest.java | 150 ++++++++
.../org/apache/sling/jobs/impl/QueueEntry.java | 48 +++
.../java/org/apache/sling/jobs/impl/UtilsTest.java | 114 ++++++
.../jobs/impl/storage/InMemoryJobStorageTest.java | 90 +++++
36 files changed, 3954 insertions(+)
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..7d34637
--- /dev/null
+++ b/README.md
@@ -0,0 +1,27 @@
+# Sling Jobs, not to be confused with Sling Event bundle.
+
+This Bundle provides Job processing using a message queue. It aims to do that using off the shelf queues. It does not
+provide an API that allows interaction with the queue beyond that is supported by the ISO AMQP standard or the JMS API. Although the
+classes here may have similar names to the API in org.apache.sling.egent.jobs, they are not the same. Methods present
+in org.apache.sling.egent.jobs API that are not compatible with a distributed message queue concept are not included, and the
+API is designed with message passing in mind.
+
+
+# How it works.
+
+MoM Queues transport messages that control the state of a Job taking it through a lifecycle from creation to distruction. Not all messages
+are required to live that lifecycle, only the create message is required under normal circumstances.
+
+A MoM QueueReader implemented by the JobQueueConsumerFactory consumes messages from MoM Queues. This is a OSGi Configuration Factory
+allowing many JobQueueConsumers to dequeue messages from a MoM Queue. When a message is dequeued and identified correctly as a
+Job related message, it is forward to the JobSubsystem component.
+
+The JobSubsystem component allows JobConsumers to register with it. It interprets the messages, creating Jobs and managing their lifecycle.
+It will identify JobConsumers that can consume the job and invoke the execute method on the JobConsumer.
+
+JobConsumers are where the job itself is implemented. Components that implement this API are registered as services which get
+registered by OSGi DS with the JobSubsystem. Each JobConsumer declares a list of JobTypes it will consume. There is a special
+JobType of ANY_TYPE for consumers that want to consume any Job. To use the JobSubsystem you will need to implement your own
+JobConsumers. There are examples of JobConsumers in this source code area.
+
+
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..2343918
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,142 @@
+<?xml version="1.0" encoding="ISO-8859-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
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>sling</artifactId>
+ <version>26</version>
+ <relativePath />
+ </parent>
+
+ <artifactId>org.apache.sling.jobs</artifactId>
+ <packaging>bundle</packaging>
+ <version>0.0.1-SNAPSHOT</version>
+
+ <name>Apache Sling Jobs Support</name>
+ <description>
+ Support for Job processing using message passing..
+ </description>
+
+ <scm>
+ <connection>scm:svn:http://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core</connection>
+ <developerConnection>scm:svn:https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core</developerConnection>
+ <url>http://svn.apache.org/viewvc/sling/trunk/contrib/commons/mom/jobs/core</url>
+ </scm>
+
+ <properties>
+ <site.jira.version.id>12315369</site.jira.version.id>
+ <sling.java.version>7</sling.java.version>
+ <exam.version>4.4.0</exam.version>
+ <url.version>2.4.5</url.version>
+ <bundle.build.dir>${basedir}/target</bundle.build.dir>
+ <bundle.file.name>${bundle.build.dir}/${project.build.finalName}.jar</bundle.file.name>
+ <min.port>37000</min.port>
+ <max.port>37999</max.port>
+ </properties>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-scr-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ </plugin>
+ </plugins>
+ </build>
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <configuration>
+ <excludePackageNames>
+ </excludePackageNames>
+ </configuration>
+ </plugin>
+ </plugins>
+ </reporting>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>org.apache.sling.mom</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>org.apache.felix.scr.annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>osgi.core</artifactId>
+ <version>6.0.0</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <version>2.0.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <!-- Use the version being used in Sling -->
+ <version>15.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>org.apache.sling.commons.osgi</artifactId>
+ <version>2.4.0</version>
+ </dependency>
+ <!-- Testing -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.service.cm</artifactId>
+ <version>1.5.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>1.10.19</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/src/main/java/org/apache/sling/jobs/Job.java b/src/main/java/org/apache/sling/jobs/Job.java
new file mode 100644
index 0000000..819a7a2
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/Job.java
@@ -0,0 +1,160 @@
+/*
+ * 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.jobs;
+
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.sling.mom.TopicManager;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Largely based on org.apache.sling.event.Job.
+ * A Job contains the complete state of the Job as known in the current JVM.
+ */
+public interface Job {
+
+
+
+ /**
+ * The current job state.
+ */
+ enum JobState {
+ QUEUED, // waiting in queue after adding or for restart after failing
+ ACTIVE, // job is currently in processing
+ SUCCEEDED, // processing finished successfully
+ STOPPED, // processing was stopped by a user
+ GIVEN_UP, // number of retries reached
+ ERROR, // processing signaled CANCELLED or throw an exception
+ ANY_STATE, // Any active state.
+ CREATED, DROPPED // dropped jobs
+ }
+
+ /**
+ * The job queue. - immutable.
+ * @return The job queue name
+ */
+ @Nonnull
+ Types.JobQueue getQueue();
+
+ /**
+ * Unique job ID. immutable.
+ * @return The unique job ID.
+ */
+ @Nonnull
+ String getId();
+
+
+ /**
+ * The job type. A job has a type, which is used by the Job Manager to offer up to JobConsumers implementing a JobTypeValve
+ * to accept.
+ * @return the job type.
+ */
+ @Nonnull
+ Types.JobType getJobType();
+
+ /**
+ * @return a map of all properties.
+ */
+ @Nonnull
+ Map<String,Object> getProperties();
+
+ /**
+ * On first execution the value of this property is zero.
+ * This property is managed by the job handling.
+ */
+ int getRetryCount();
+
+ /**
+ * The property to track the retry maximum retry count for jobs.
+ * This property is managed by the job handling.
+ */
+ int getNumberOfRetries();
+
+
+ /**
+ * The time when the job started.
+ */
+ long getStarted();
+
+ /**
+ * @return The time when the job was created.
+ */
+ long getCreated();
+
+
+ /**
+ * Get the job state
+ */
+ @Nonnull
+ JobState getJobState();
+
+ /**
+ * Set the new state.
+ * @param newState
+ */
+ void setState(@Nonnull JobState newState);
+
+
+ /**
+ * If the job is cancelled or succeeded, this method will return the finish date.
+ * @return The finish date or <code>null</code>
+ */
+ long getFinished();
+
+ /**
+ * This method returns the message from the last job processing, regardless
+ * whether the processing failed, succeeded or was cancelled. The message
+ * is optional and can be set by a job consumer.
+ * @return The result message or <code>null</code>
+ */
+ @Nullable
+ String getResultMessage();
+
+ @Nonnull
+ JobUpdateBuilder newJobUpdateBuilder();
+
+
+ /**
+ * @return the controller associated with this Job, if one is present in the current JVM otherwise null.
+ */
+ @Nullable
+ JobController getController();
+
+
+ /**
+ * Set the current job controller.
+ */
+ void setJobController(@Nonnull JobController jobController);
+
+
+ /**
+ * Remove the current job controller.
+ */
+ void removeJobController();
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobBuilder.java b/src/main/java/org/apache/sling/jobs/JobBuilder.java
new file mode 100644
index 0000000..b69934d
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobBuilder.java
@@ -0,0 +1,44 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+import java.util.Map;
+
+/**
+ * A JobBuilder allows users of the JobSystem to modify the properties of the Job and submit it for processing.
+ */
+public interface JobBuilder {
+
+ /**
+ * Set the optional configuration properties for the job.
+ * @param props The properties of the job. All values must be {@code java.io.Serializable}.
+ * @return The job builder to continue building.
+ */
+ @Nonnull
+ JobBuilder addProperties(@Nonnull Map<String, Object> props);
+
+ /**
+ * Add the job for processing.
+ * @return The job or <code>null</code>
+ */
+ @Nonnull
+ Job add();
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobCallback.java b/src/main/java/org/apache/sling/jobs/JobCallback.java
new file mode 100644
index 0000000..bbadc6b
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobCallback.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jobs;
+
+/**
+ * When a job is complete, the callback gets called.
+ * This interface is implemented by the Jobs Implementation and passed into a JobsConsumer.
+ */
+public interface JobCallback {
+
+ /**
+ * Call back with an update job state.
+ * @param finalJobState
+ */
+ void callback(Job finalJobState);
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobConsumer.java b/src/main/java/org/apache/sling/jobs/JobConsumer.java
new file mode 100644
index 0000000..dbf6703
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobConsumer.java
@@ -0,0 +1,46 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Components that implement JobConsumers are registered with the Job Sub System by an OSGi Whiteboard pattern. Once
+ * registered the component will be offered jobs matching the job types set as the OSGi property JobConsumer.JOB_TYPES.
+ * If the consumer accepts the job, by not throwing and exception on the execute method, it agrees to ensure that the job
+ * is executed. It may queue the job, but if queued it must ensure the job is executed. A JobConsumer should check that he
+ * dependencies required to execute the job are satisfied before accepting the job for execution. Queueing jobs in the job consumer
+ * to ensure that the execution threads of the JobConsumer have work available is ok, but queuing jobs to remove them from
+ * the job queue is to be discouraged, as it may require too much from the JobConsumer implementation to guarantee the contract
+ * in this interface.
+ */
+public interface JobConsumer {
+
+ String JOB_TYPES = "job.types";
+
+ /**
+ * Execute the job, given an initial state and supply a JobStateListener to allow the job to send updates of state change
+ * @param initialState the initial state of the Job.
+ * @param listener a listener to propagate JobUpdates.
+ * @param callback called when the job is completed.
+ * @throws RuntimeException or any subclass when the Job offered in initialState cant be accepted for execution.
+ */
+ @Nonnull
+ void execute(@Nonnull Job initialState, @Nonnull JobUpdateListener listener, @Nonnull JobCallback callback);
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobController.java b/src/main/java/org/apache/sling/jobs/JobController.java
new file mode 100644
index 0000000..e038b52
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobController.java
@@ -0,0 +1,43 @@
+/*
+ * 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.jobs;
+
+/**
+ * A job controller provides a mechanism by which a running job can be sent control messages.
+ * Jobs when they start should register a JobController implementation with the JobManager so that
+ * it can act on control messages sent. Alternatively a TopicListener can be used to route
+ * message to the Job, bypassing the manager, although this TopicListener will need to filter
+ * for all messages associated with the Job.
+ */
+public interface JobController {
+
+
+ /**
+ * Stop the job as soon as appropriate.
+ */
+ void stop();
+
+ /**
+ * Abort the job immediately.
+ */
+ void abort();
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobManager.java b/src/main/java/org/apache/sling/jobs/JobManager.java
new file mode 100644
index 0000000..5dc4c3c
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobManager.java
@@ -0,0 +1,165 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Based on org.apache.sling.event.job.JobManager but trimmed down and adjusted to deal with distributed job queues.
+ */
+public interface JobManager {
+
+
+
+
+
+ /**
+ * Fluent API to create, start and schedule new jobs
+ * @param queue Required queue. This job will be submitted to the named queue.
+ * @param jobType required job type. The job will have a job type which is used to select the Job consumer that can
+ * process the job.
+ * @return A job builder
+ */
+ @Nonnull
+ JobBuilder newJobBuilder(@Nonnull Types.JobQueue queue, @Nonnull Types.JobType jobType);
+
+ /**
+ * 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>
+ */
+
+ @Nullable
+ Job getJobById(@Nonnull String jobId);
+
+
+ /**
+ * This method comes from the existing Sling Event API and has not been implemented, as it was found
+ * not to be necessary to implement Jobs on the MoM API. Since it was not implemented
+ * It has not been included in the first release of the API.
+ * If it is required, it can be added and implemented.
+ *
+ * ---------------------------------------
+ * Find a job - either queued or active.
+ *
+ * This method searches for a job 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.
+ *
+ * Even if the method return null, the Job may exist remotely and has not yet sent any messages.
+ *
+ * @param queue Topic is required.
+ * @param template The map acts like a template. The searched job
+ * must match the template (AND query).
+ * @return A job or <code>null</code>
+ * @Nullable
+ * Job getJob(@Nonnull Types.JobQueue queue, @Nonnull Map<String, Object> template);
+ *
+ */
+
+ /**
+ * The requested job types for the query.
+ * This can either be all (unfinished) jobs, all activated (started) or all queued jobs.
+ */
+ enum QueryType {
+ ALL, // all means all active and all queued
+ ACTIVE,
+ QUEUED,
+ HISTORY, // returns the complete history of cancelled and succeeded jobs (if available)
+ CANCELLED, // history of cancelled jobs (STOPPED, GIVEN_UP, ERROR, DROPPED)
+ SUCCEEDED, // history of succeeded jobs
+ STOPPED, // history of stopped jobs
+ GIVEN_UP, // history of given up jobs
+ ERROR, // history of jobs signaled CANCELLED or throw an exception
+ DROPPED // history of dropped jobs
+ }
+ /**
+ * This method comes from the existing Sling Event API and has not been implemented, as it was found
+ * not to be necessary to implement Jobs on the MoM API. Since it was not implemented
+ * It has not been included in the first release of the API.
+ * If it is required, it can be added and implemented.
+ *
+ * ---------------------------------------
+ * Return all jobs of a given type.
+ *
+ * Based on the type parameter, either the history of jobs can be returned or unfinished jobs. The type
+ * parameter can further specify which category of jobs should be returned: for the history either
+ * succeeded jobs, cancelled jobs or both in combination can be returned. For unfinished jobs, either
+ * queued jobs, started jobs or the combination can be returned.
+ * 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.
+ *
+ * The Jobs returned are only jobs known to the JobManager instance.
+ *
+ * @param type Required parameter for the type. See above.
+ * @param queue Queue Name 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
+ * of zero or less indicates that all jobs should be returned.
+ * @param templates A list of filter property maps. Each map acts like a template. The searched job
+ * must match the template (AND query). By providing several maps, different filters
+ * are possible (OR query).
+ * @return A collection of jobs - the collection might be empty.
+ * @Nonnull
+ * Collection<Job> findJobs(@Nonnull QueryType type, @Nonnull Types.JobQueue queue, long limit, @Nullable Map<String, Object>... templates);
+ */
+
+ /**
+ * Stop a job, by sending a message.
+ * When a job is stopped and the job consumer supports stopping the job processing, it is up
+ * to the job consumer how the stopping is handled. The job can be marked as finished successful,
+ * permanently failed or being retried.
+ */
+ void stopJobById(@Nonnull String jobId);
+
+ /**
+ * Aborts a job by sending an abort message, may or may not be successful.
+ *
+ *
+ * @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> if the job is not know to the job manager, however the abort message should be sent anyway.
+ */
+ boolean abortJob(@Nonnull String jobId);
+
+
+ /**
+ * Retry a cancelled job, by requeuing.
+ * If a job has failed permanently it can be requeued with this method. The job will be
+ * removed from the history and put into the queue again. The new job will get a new job id.
+ * For all other jobs calling this method has no effect and it simply returns <code>null</code>.
+ * @param jobId The job id.
+ * @return If the job is requeued, the new job object otherwise <code>null</code>
+ */
+ @Nullable
+ Job retryJobById(@Nonnull String jobId);
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobTypeValve.java b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
new file mode 100644
index 0000000..20f6782
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
@@ -0,0 +1,36 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+
+/**
+ * An interface to allow a component, normally a JobConsumer to inspect a JobType and indicate that
+ * it can perform further actions on it. JobTypeValves are used in place of static JobType declarations.
+ * This interface is only implemented where static declaration of JobTypes does not satisfy the use cases.
+ */
+public interface JobTypeValve {
+
+ /**
+ * Return true if the component, normally a JobConsumer, can process the jobType.
+ * @param jobType
+ * @return true if can be processed.
+ */
+ boolean accept(@Nonnull Types.JobType jobType);
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdate.java b/src/main/java/org/apache/sling/jobs/JobUpdate.java
new file mode 100644
index 0000000..62723ad
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobUpdate.java
@@ -0,0 +1,162 @@
+/*
+ * 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.jobs;
+
+import org.apache.sling.mom.TopicManager;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+
+/**
+ * Job updates are messages that update the state of the job at all subscribers.
+ * The distribution mechanism for JobUpdate messages must ensure that the messages are distributed appropriately so that
+ * the queues that contain the messages allow the messages to be acted on in an appropriate manner. For instance if the
+ * implementation decides that it will use a message queue to queue jobs as JobUpdate messages, then it may need to
+ * put the JobUpdates that start a message on a separate queue from the JobUpdates that update, abort or stop a job, so that
+ * those messages do not get delayed by a backlog of job start messages. Alternatively the implementation may decide to implement
+ * a scheduler that consumes job start messages at full queue throughput and queue jobs into a separate implementation specific queue.
+ * <p/>
+ * The API does not specify the implementation, only that the various JobUpdate messages are delivered with an delay appropriate for the
+ * type of message, identified by the JobUpdateCommand type.
+ */
+public interface JobUpdate {
+
+ /**
+ * The time of the update. The implementation may choose to adjust its concept of time to
+ * ensure the that JobUpdates it emits guaranteed to be applied in the correct order. How
+ * it achieves this is an implementation detail. Consumers of JobUpdates should define a policy
+ * on how updates are applied. Out of sequence updates may or may not be applied to a job, depending on
+ * the implementation.
+ * @return the timestamp of the update.
+ */
+ long updateTimestamp();
+
+
+ /**
+ * Update messages may have a time to live. Where the update command is abort or stop and the job message doesn't
+ * exist on the receiver the update message may be stored until the TTL expires before it is applied. This allows
+ * messages sent out of order to wait till expires has been reached before applying the update. The rules determining
+ * how expiring messages are applied is an implementation detail.
+ * @return
+ */
+ long expires();
+
+ /**
+ * The Job type. A job type is used to identify the type of job and hence the JobConsumer (via JobTypeValve) that can
+ * execute the Job.
+ * @return the job type.
+ */
+ @Nonnull
+ Types.JobType getJobType();
+
+ enum JobUpdateCommand {
+ START_JOB, // Starts the job.
+ UPDATE_JOB, // Update the job.
+ ABORT_JOB, // Abort the job, if the receiver is running the job, then it must abort it.
+ RETRY_JOB, // Retry the job.
+ STOP_JOB // Stop the job from running, but if already running let the job complete
+ ;
+
+ public org.apache.sling.mom.Types.CommandName asCommandName() {
+ return org.apache.sling.mom.Types.commandName(toString());
+ }
+ }
+
+
+ /**
+ * @return the type of update.
+ */
+ @Nonnull
+ JobUpdateCommand getCommand();
+
+
+ /**
+ * The job queue. - immutable.
+ * @return The job queue
+ */
+ @Nonnull
+ Types.JobQueue getQueue();
+
+ /**
+ * Unique job ID. immutable.
+ * @return The unique job ID.
+ */
+ @Nonnull
+ String getId();
+
+ /**
+ *
+ * @return the new state of the job, may not have changed from the old state.
+ */
+ @Nonnull
+ Job.JobState getState();
+
+ enum JobPropertyAction {
+ REMOVE
+ }
+
+ /**
+ * The changed properties of the job. A property value of JobPropertyAction.REMOVE causes the property to be removed.
+ * Property types cant be changed in 1 message.
+ * @return the map of property changes, will be <code>null</code> if the update message does not contain property changes.
+ */
+ @Nonnull
+ Map<String, Object> getProperties();
+
+ /**
+ * On first execution the value of this property is zero.
+ * This property is managed by the job handling.
+ */
+ int getRetryCount();
+
+ /**
+ * The property to track the retry maximum retry count for jobs.
+ * This property is managed by the job handling.
+ */
+ int getNumberOfRetries();
+
+
+ /**
+ * The time when the job started.
+ */
+ long getStarted();
+
+ /**
+ * @return The time when the job was created.
+ */
+ long getCreated();
+
+
+ /**
+ * If the job is cancelled or succeeded, this method will return the finish date.
+ * @return The finish date or <code>null</code>
+ */
+ long getFinished();
+
+ /**
+ * This method returns the message from the last job processing, regardless
+ * whether the processing failed, succeeded or was cancelled. The message
+ * is optional and can be set by a job consumer.
+ * @return The result message or <code>null</code>
+ */
+ @Nullable
+ String getResultMessage();
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java b/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java
new file mode 100644
index 0000000..d691348
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java
@@ -0,0 +1,48 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+
+/**
+ * Create a job update ading properties and building.
+ */
+public interface JobUpdateBuilder {
+
+ @Nonnull
+ JobUpdateBuilder command(@Nonnull JobUpdate.JobUpdateCommand command);
+
+ @Nonnull
+ JobUpdate build();
+
+ /**
+ * Set a property to update.
+ * @param name the name of the property
+ * @param value the value of the property which may be null. To remove the property set the value to JobUpdate.JobPropertyAction.REMOVE.
+ * @return this JobBuilder instance.
+ */
+ @Nonnull
+ JobUpdateBuilder put(@Nonnull String name, @Nullable Object value);
+
+ @Nonnull
+ JobUpdateBuilder putAll(@Nonnull Map<String, Object> properties);
+}
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdateListener.java b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
new file mode 100644
index 0000000..82a2c63
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
@@ -0,0 +1,34 @@
+/*
+ * 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.jobs;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Listens for JobUpdate messages.
+ * Examples of a JobUpdateListener include implementations of a Job that consumes JobUpdateMessages, or JobConsumers that execute a Job.
+ */
+public interface JobUpdateListener {
+
+ /**
+ * Update the job state with a message.
+ * @param update
+ */
+ void update(@Nonnull JobUpdate update);
+}
diff --git a/src/main/java/org/apache/sling/jobs/Types.java b/src/main/java/org/apache/sling/jobs/Types.java
new file mode 100644
index 0000000..00f6501
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/Types.java
@@ -0,0 +1,183 @@
+/*
+ * 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.jobs;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * This has to be part of the API to prevent all sorts of other issues.
+ */
+public final class Types {
+
+
+
+
+ private Types() {
+
+ }
+
+ /**
+ * Interface to identify JobQueues. JobQueues have an associated MoM Queue and a MoM Topic, both of which
+ * are available via this interface. The MoM Queue contains commands, the the MoM Topic contains status updates.
+ */
+ public interface JobQueue {
+ org.apache.sling.mom.Types.QueueName asQueueName();
+
+ org.apache.sling.mom.Types.TopicName asTopicName();
+
+ }
+
+ /**
+ * Interface to identify JobTypes. Jobs are identified by a JobType, independent of the JobQueue or the underlying
+ * MoM Queue or MoM Topic.
+ */
+ public interface JobType {
+ }
+
+ /**
+ * Create a JobQueue from a string represtnation, normally this us a path representing the jobQueue, although the path
+ * hierachy may not have any significance depending on the MoM implementation.
+ * @param jobQueue
+ * @return
+ */
+ public static JobQueue jobQueue(String jobQueue) {
+ return new JobQueueImpl(jobQueue);
+ }
+
+ /**
+ * Create a JobType from a string representation.
+ * @param jobType
+ * @return
+ */
+ public static JobType jobType(String jobType) {
+ return new JobTypeImpl(jobType);
+ }
+
+ /**
+ * A special JobQueue that represents any available queue. Normally used for filtering queue messages. Probably wont
+ * work as a destination queue (broadcast) depending on the MoM Implementation.
+ */
+ public static JobQueue ANY_JOB_QUEUE = new AnyJobQueue();
+
+
+ public static Set<JobType> jobType(String[] types) {
+ Set<JobType> hs = new HashSet<JobType>();
+ for ( String s : types) {
+ hs.add(jobType(s));
+ }
+ return Collections.unmodifiableSet(hs);
+ }
+
+
+
+
+ /**
+ * Wraps a JobType.
+ */
+ private static class JobTypeImpl extends StringWrapper implements JobType {
+
+ private JobTypeImpl(String jobType) {
+ super(jobType);
+ }
+ }
+
+
+
+ /**
+ * Wraps a JobQueue.
+ */
+ private static class JobQueueImpl extends StringWrapper implements JobQueue {
+
+ private JobQueueImpl(String jobQueue) {
+ super(jobQueue);
+ }
+
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == ANY_JOB_QUEUE || super.equals(obj);
+ }
+
+
+ @Override
+ public org.apache.sling.mom.Types.QueueName asQueueName() {
+ return org.apache.sling.mom.Types.queueName(toString());
+ }
+
+ @Override
+ public org.apache.sling.mom.Types.TopicName asTopicName() {
+ return org.apache.sling.mom.Types.topicName(toString());
+ }
+ }
+
+ /**
+ * Special JobQueue to match any.
+ */
+ private static class AnyJobQueue extends JobQueueImpl {
+
+
+ private AnyJobQueue() {
+ super("*");
+ }
+
+
+ @Override
+ public boolean equals(Object obj) {
+ return true;
+ }
+
+ @Override
+ public int compareTo(String o) {
+ return 0;
+ }
+ }
+
+ private static class StringWrapper implements Comparable<String> {
+
+
+ private String value;
+
+ private StringWrapper(String value) {
+ this.value = value;
+ }
+
+ @Override
+ public int hashCode() {
+ return value.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return value.equals(obj.toString());
+ }
+
+ @Override
+ public int compareTo(String o) {
+ return value.compareTo(o);
+ }
+
+ public String toString() {
+ return value;
+ }
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobBuilderImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobBuilderImpl.java
new file mode 100644
index 0000000..a9da26a
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobBuilderImpl.java
@@ -0,0 +1,68 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobBuilder;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.jobs.impl.spi.JobStarter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Provides an implementation of a JobBuilder.
+ */
+public class JobBuilderImpl implements JobBuilder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobBuilderImpl.class);
+ private final String id;
+ private final Map<String, Object> properties;
+ private final JobStarter jobStarter;
+ private final Types.JobQueue topic;
+ private final Types.JobType jobType;
+
+
+ public JobBuilderImpl(JobStarter jobStarter, Types.JobQueue topic, Types.JobType jobType) {
+ this.jobStarter = jobStarter;
+ this.topic = topic;
+ this.jobType = jobType;
+ this.id = Utils.generateId();
+ properties = new HashMap<String, Object>();
+ }
+
+
+ @Nonnull
+ @Override
+ public JobBuilder addProperties(@Nonnull Map<String, Object> props) {
+ this.properties.putAll(props);
+ return this;
+ }
+
+ @Nonnull
+ @Override
+ public Job add() {
+ return jobStarter.start(new JobImpl(topic, id, jobType, properties));
+ }
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobImpl.java
new file mode 100644
index 0000000..ebc0971
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobImpl.java
@@ -0,0 +1,244 @@
+/*
+ * 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.jobs.impl;
+
+import com.google.common.base.Preconditions;
+import org.apache.sling.jobs.*;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * JobImpl is a data object to hold the current state of the job in the current JVM as loaded into memory.
+ * The JobImpl also listens for JobUpdates.
+ */
+public class JobImpl implements Job, JobUpdateListener {
+ private final Types.JobQueue jobQueue;
+ private final String id;
+ private final Map<String, Object> properties = new HashMap<String, Object>();
+ private int retryCount;
+ private int numberOfRetries;
+ private long startedAt;
+ private long createdAt;
+ private long lastUpdate = 0;
+ private JobState jobState;
+ private long finishedAt;
+ private String resultMessage;
+ private JobController jobController;
+ private Types.JobType jobType;
+
+ public JobImpl(@Nonnull Types.JobQueue jobQueue, @Nonnull String id, @Nonnull Types.JobType jobType, @Nonnull Map<String, Object> properties) {
+ this.jobQueue = jobQueue;
+ this.jobType = jobType;
+ this.id = id;
+ this.resultMessage = "";
+ this.createdAt = System.currentTimeMillis();
+ this.jobState = JobState.CREATED;
+ this.properties.putAll(properties);
+ }
+
+ public JobImpl(JobUpdate update) {
+ this.jobQueue = update.getQueue();
+ this.id = update.getId();
+ update(update);
+ updateProperties(update.getProperties());
+ }
+
+ @Nonnull
+ @Override
+ public Types.JobQueue getQueue() {
+ return jobQueue;
+ }
+
+ @Nonnull
+ @Override
+ public String getId() {
+ return id;
+ }
+
+ @Nonnull
+ @Override
+ public Types.JobType getJobType() {
+ return jobType;
+ }
+
+ @Nonnull
+ @Override
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+
+ @Override
+ public int getRetryCount() {
+ return retryCount;
+ }
+
+ @Override
+ public int getNumberOfRetries() {
+ return numberOfRetries;
+ }
+
+ @Override
+ public long getStarted() {
+ return startedAt;
+ }
+
+ @Override
+ public long getCreated() {
+ return createdAt;
+ }
+
+ @Nonnull
+ @Override
+ public JobState getJobState() {
+ return jobState;
+ }
+
+ @Override
+ public void setState(@Nonnull JobState newState) {
+ jobState = newState;
+ }
+
+ @Override
+ public long getFinished() {
+ return finishedAt;
+ }
+
+ @Nullable
+ @Override
+ public String getResultMessage() {
+ return resultMessage;
+ }
+
+
+ @Nullable
+ @Override
+ public JobController getController() {
+ return jobController;
+ }
+
+ @Override
+ public void setJobController(@Nonnull JobController jobController) {
+ this.jobController = jobController;
+ }
+
+ @Override
+ public void removeJobController() {
+ jobController = null;
+ }
+
+ /**
+ * Apply an job update to this job, checking that the update is valid for the job.
+ * @param jobUpdate
+ */
+ @Override
+ public void update(@Nonnull JobUpdate jobUpdate) {
+ if ( id.equals(jobUpdate.getId()) && ( jobQueue == Types.ANY_JOB_QUEUE || jobQueue.equals(jobUpdate.getQueue()))) {
+ // Start Job commands always go onto a queue and dont expire.
+ if ( jobUpdate.getCommand() != JobUpdate.JobUpdateCommand.START_JOB && jobUpdate.expires() < System.currentTimeMillis()) {
+ throw new IllegalStateException(
+ "JobUpdate has expired, can't be applied. Expired at "+jobUpdate.expires()+
+ ", time now "+System.currentTimeMillis()+
+ " expired "+(System.currentTimeMillis()-jobUpdate.expires())+" ms ago.");
+ }
+ if (jobUpdate.updateTimestamp() < lastUpdate ) {
+ throw new IllegalStateException("JobUpdate received out of sequence, cant be applied. Last Update was at "+lastUpdate+" this update is at "+jobUpdate.updateTimestamp());
+ }
+ lastUpdate = jobUpdate.updateTimestamp();
+ switch(jobUpdate.getCommand()) {
+ case START_JOB:
+ updateState(jobUpdate);
+ updateProperties(jobUpdate.getProperties());
+ break;
+ case UPDATE_JOB:
+ // note, when job first comes into existence it is updated, then started.
+ // the start message is a queued message, the update is a jobQueue or pub sub message.
+ updateState(jobUpdate);
+ updateProperties(jobUpdate.getProperties());
+ break;
+ case RETRY_JOB:
+ updateState(jobUpdate);
+ // Allow more retries.
+ numberOfRetries = retryCount + numberOfRetries;
+ // TODO: trigger retry if required.
+ updateProperties(jobUpdate.getProperties());
+ break;
+ case STOP_JOB:
+ if (jobController != null) {
+ jobController.stop();
+ }
+ break;
+ case ABORT_JOB:
+ if (jobController != null) {
+ jobController.abort();
+ }
+ break;
+ }
+ } else {
+ throw new IllegalArgumentException("Cant update job with jobUpdate that doesn't match id and jobQueue ");
+ }
+ }
+
+ /**
+ * Update the properties taking into account any PropertyActions required.
+ * @param properties the update properties.
+ */
+ private void updateProperties(@Nonnull Map<String, Object> properties) {
+ Preconditions.checkNotNull(properties, "Properties cant be null.");
+ for (Map.Entry<String, Object> e : properties.entrySet()) {
+ if (e.getValue() instanceof JobUpdate.JobPropertyAction ) {
+ switch(((JobUpdate.JobPropertyAction)e.getValue())) {
+ case REMOVE:
+ this.properties.remove(e.getKey());
+ break;
+ }
+ } else {
+ this.properties.put(e.getKey(), e.getValue());
+ }
+ }
+ }
+
+ /**
+ * Update the jobstate data for the job.
+ * @param jobUpdate
+ */
+ private void updateState(@Nonnull JobUpdate jobUpdate) {
+ retryCount = jobUpdate.getRetryCount();
+ jobType = jobUpdate.getJobType();
+ numberOfRetries = jobUpdate.getNumberOfRetries();
+ startedAt = jobUpdate.getStarted();
+ createdAt = jobUpdate.getCreated();
+ finishedAt = jobUpdate.getFinished();
+ resultMessage = jobUpdate.getResultMessage();
+ jobState = jobUpdate.getState();
+ }
+
+ /**
+ * Get a JobUpdateBuilder for this Job.
+ * @return the job update builder.
+ */
+ @Nonnull
+ @Override
+ public JobUpdateBuilder newJobUpdateBuilder() {
+ return new JobUpdateBuilderImpl(this);
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java
new file mode 100644
index 0000000..5d9ff4b
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java
@@ -0,0 +1,123 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.*;
+import org.apache.sling.jobs.impl.spi.JobStarter;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Implements a JobManager, storing Jobs in a JobStorage implementation.
+ * Implements a JobStarter so it can start Jobs and queue them by sending a message.
+ * Implements a JobUpdateListener so it can store inbound update messages.
+ * Requires a JobStorage implementation to store jobs and a JobListener to process update messages.
+ * Requires a JobUpdateLister which it uses to send JobUpdates to.
+ * Does not run jobs, that is performed by implementations of JobConsumer listening to queues.
+ */
+public class JobManagerImpl implements JobManager, JobStarter, JobUpdateListener {
+
+
+ /**
+ * Storage for jobs.
+ */
+ private final JobStorage jobStorage;
+ /**
+ * A listener for job updates.
+ */
+ private final JobUpdateListener messageSender;
+
+ /**
+ *
+ * @param jobStorage
+ * @param messageSender
+ */
+ public JobManagerImpl(JobStorage jobStorage, JobUpdateListener messageSender) {
+ this.jobStorage = jobStorage;
+ this.messageSender = messageSender;
+ }
+
+ @Nonnull
+ @Override
+ public JobBuilder newJobBuilder(@Nonnull Types.JobQueue queue, @Nonnull Types.JobType jobType) {
+ return new JobBuilderImpl(this, queue, jobType);
+ }
+
+ @Nullable
+ @Override
+ public Job getJobById(@Nonnull String jobId) {
+ return jobStorage.get(jobId);
+ }
+
+ @Override
+ public void stopJobById(@Nonnull String jobId) {
+ Job job = getJobById(jobId);
+ if ( job != null) {
+ messageSender.update(job.newJobUpdateBuilder().command(JobUpdate.JobUpdateCommand.STOP_JOB).build());
+ }
+ }
+
+ @Override
+ public boolean abortJob(@Nonnull String jobId) {
+ Job job = getJobById(jobId);
+ if ( job != null) {
+ messageSender.update(job.newJobUpdateBuilder().command(JobUpdate.JobUpdateCommand.ABORT_JOB).build());
+ return true;
+ } else {
+ messageSender.update(new JobUpdateBuilderImpl(jobId).command(JobUpdate.JobUpdateCommand.ABORT_JOB).build());
+ return false;
+ }
+ }
+
+ @Nullable
+ @Override
+ public Job retryJobById(@Nonnull String jobId) {
+ Job job = getJobById(jobId);
+ if (job != null) {
+ messageSender.update(job.newJobUpdateBuilder().command(JobUpdate.JobUpdateCommand.RETRY_JOB).build());
+ }
+ return job;
+ }
+
+
+ @Override
+ public Job start(Job j) {
+ j.setState(Job.JobState.QUEUED);
+ messageSender.update(j.newJobUpdateBuilder().command(JobUpdate.JobUpdateCommand.START_JOB).putAll(j.getProperties()).build());
+ return jobStorage.put(j);
+ }
+
+
+ @Override
+ public void update(@Nonnull JobUpdate update) {
+ Job j = jobStorage.get(update.getId());
+ if ( j instanceof JobUpdateListener ) {
+ ((JobUpdateListener) j).update(update);
+ } else {
+ jobStorage.put(new JobImpl(update));
+ }
+ }
+
+ public void dispose() {
+ }
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobQueueConsumerFactory.java b/src/main/java/org/apache/sling/jobs/impl/JobQueueConsumerFactory.java
new file mode 100644
index 0000000..a1091ec
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobQueueConsumerFactory.java
@@ -0,0 +1,114 @@
+/*
+ * 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.jobs.impl;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.felix.scr.annotations.*;
+import org.apache.sling.jobs.*;
+import org.apache.sling.mom.*;
+import org.apache.sling.mom.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+/**
+ * This is a configuration factory that creates QueueReader instances on configuration. These connect to the JobManager
+ * service and are registered using the OSGi Whiteboard pattern with the QueueManager. The JobManager service must implement JobConsumer.
+ *
+ */
+@Component(configurationFactory = true,
+ policy = ConfigurationPolicy.REQUIRE,
+ metatype = true,
+ immediate = true)
+@Properties({
+ @Property(name= QueueReader.QUEUE_NAME_PROP)
+})
+@Service(value = QueueReader.class)
+public class JobQueueConsumerFactory implements QueueReader, MessageFilter {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobQueueConsumerFactory.class);
+ private static final Set<JobUpdate.JobUpdateCommand> ALLOWED_COMMANDS = ImmutableSet.of(JobUpdate.JobUpdateCommand.UPDATE_JOB) ;
+
+ @Reference
+ private JobManager jobManager;
+
+ @Reference
+ private TopicManager topicManager;
+ @Reference
+ private QueueManager queueManager;
+
+ @Activate
+ public void activate(Map<String, Object> properties) {
+ if ( !(jobManager instanceof JobConsumer) ) {
+ LOGGER.error("JobManager must implement JobConsumer interface. {} does not. ", jobManager.getClass());
+ throw new IllegalStateException("JobManager does not implement JobConsumer");
+ }
+ }
+
+ @Deactivate
+ public void deactivate(@SuppressWarnings("UnusedParameters") Map<String, Object> properties) {
+ }
+
+
+
+
+ @Override
+ public void onMessage(Types.QueueName queueName, Map<String, Object> message) throws RequeueMessageException {
+
+ final Job job = new JobImpl(new JobUpdateImpl(message));
+
+
+ ((JobConsumer)jobManager).execute(job, new JobUpdateListener() {
+ @Override
+ public void update(@Nonnull JobUpdate update) {
+ if (update.getId() != job.getId() || !ALLOWED_COMMANDS.contains(update.getCommand())) {
+
+ throw new IllegalArgumentException("Not allowed to update other jobs or issue reserved commands when updating the state of a running job.");
+ }
+ topicManager.publish(update.getQueue().asTopicName(), update.getCommand().asCommandName(), Utils.toMapValue(update));
+ }
+ }, new JobCallback() {
+ @Override
+ public void callback(Job finalJobState) {
+ if (finalJobState.getId() != job.getId()) {
+ throw new IllegalArgumentException("Final Job state ID must match initial JobState ID");
+ }
+ JobUpdate finalJobUpdate = finalJobState.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.UPDATE_JOB)
+ .putAll(finalJobState.getProperties())
+ .build();
+ topicManager.publish(finalJobUpdate.getQueue().asTopicName(), finalJobUpdate.getCommand().asCommandName(), Utils.toMapValue(finalJobUpdate));
+ }
+ });
+
+
+ }
+
+ @Override
+ public boolean accept(Types.Name name, Map<String, Object> mapMessage) {
+ return !(jobManager instanceof MessageFilter) || ((MessageFilter) jobManager).accept(name, mapMessage);
+ }
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java b/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java
new file mode 100644
index 0000000..ef3123b
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java
@@ -0,0 +1,199 @@
+/*
+ * 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.jobs.impl;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.ReferencePolicy;
+import org.apache.felix.scr.annotations.Service;
+import org.apache.sling.commons.osgi.PropertiesUtil;
+import org.apache.sling.jobs.*;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+import org.apache.sling.jobs.impl.storage.InMemoryJobStorage;
+import org.apache.sling.mom.QueueManager;
+import org.apache.sling.mom.TopicManager;
+import org.osgi.framework.ServiceReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * NB, this does *not* register as a JobConsumer service. it implements a JobConsumer so that it can consume Jobs from JobQueueConsumers.
+ */
+@Component(immediate = true)
+@Service(value = JobManager.class)
+public class JobSubsystem implements JobManager, JobConsumer {
+
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobSubsystem.class);
+ private JobManagerImpl manager;
+ private JobStorage jobStorage;
+ private OutboundJobUpdateListener messageSender;
+
+ /**
+ * Contains a map of JobConsumers wrapped by JobConsumerHolders keyed by ServiceReference.
+ */
+ @Reference(referenceInterface = JobConsumer.class,
+ cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE,
+ policy = ReferencePolicy.DYNAMIC,
+ bind="addConsumer",
+ unbind="removeConsumer")
+ private final Map<ServiceReference<JobConsumer>, JobConsumerHolder> registrations =
+ new ConcurrentHashMap<ServiceReference<JobConsumer>, JobConsumerHolder>();
+
+ @Reference
+ private TopicManager topicManager;
+ @Reference
+ private QueueManager queueManager;
+
+ @Activate
+ public synchronized void activate(@SuppressWarnings("UnusedParameters") Map<String, Object> properties) {
+ jobStorage = new InMemoryJobStorage();
+ messageSender = new OutboundJobUpdateListener(topicManager, queueManager);
+ manager = new JobManagerImpl(jobStorage, messageSender);
+ }
+
+ @Deactivate
+ public synchronized void deactivate(@SuppressWarnings("UnusedParameters") Map<String, Object> properties) {
+ for (Map.Entry<ServiceReference<JobConsumer>, JobConsumerHolder> e : registrations.entrySet()) {
+ e.getValue().close();
+ }
+ registrations.clear();
+ manager.dispose();
+ messageSender.dispose();
+ jobStorage.dispose();
+ }
+
+ // --- Job Manager.
+ @Nonnull
+ @Override
+ public JobBuilder newJobBuilder(@Nonnull Types.JobQueue queue, @Nonnull Types.JobType jobType) {
+ return manager.newJobBuilder(queue, jobType);
+ }
+
+ @Nullable
+ @Override
+ public Job getJobById(@Nonnull String jobId) {
+ return manager.getJobById(jobId);
+ }
+
+ @Override
+ public void stopJobById(@Nonnull String jobId) {
+ manager.stopJobById(jobId);
+ }
+
+ @Override
+ public boolean abortJob(@Nonnull String jobId) {
+ return manager.abortJob(jobId);
+ }
+
+ @Nullable
+ @Override
+ public Job retryJobById(@Nonnull String jobId) {
+ return manager.retryJobById(jobId);
+ }
+
+
+ // ---- JobConsumer Registration
+ // Register Consumers using
+ public synchronized void addConsumer(ServiceReference<JobConsumer> serviceRef) {
+ if (registrations.containsKey(serviceRef)) {
+ LOGGER.error("Registration for service reference is already present {}",serviceRef);
+ return;
+ }
+ JobConsumerHolder jobConsumerHolder = new JobConsumerHolder(serviceRef.getBundle().getBundleContext().getService(serviceRef), getServiceProperties(serviceRef));
+ registrations.put(serviceRef, jobConsumerHolder);
+ }
+
+ private Map<Object, Object> getServiceProperties(ServiceReference<JobConsumer> serviceRef) {
+ ImmutableMap.Builder<Object, Object> builder = ImmutableMap.builder();
+ for ( String k : serviceRef.getPropertyKeys()) {
+ builder.put(k, serviceRef.getProperty(k));
+ }
+ return builder.build();
+ }
+
+ public synchronized void removeConsumer(ServiceReference<JobConsumer> serviceRef) {
+ JobConsumerHolder jobConsumerHolder = registrations.remove(serviceRef);
+ if ( jobConsumerHolder != null) {
+ jobConsumerHolder.close();
+ }
+ }
+
+
+ // ------- job execution, invoked by JobQueueConsumerFactory.
+ @Nonnull
+ @Override
+ public void execute(@Nonnull Job initialState, @Nonnull JobUpdateListener listener, @Nonnull JobCallback callback) {
+ // iterate over the entries. This should cause the entries to come out in natural key order
+ // which should respect any priority applied to the Services via ServiceReference. (TODO: check that is the case)
+ // TODO: add a Job controller to the job before executing.
+ for (Map.Entry<ServiceReference<JobConsumer>,JobConsumerHolder> e : registrations.entrySet()) {
+ JobConsumerHolder jobConsumerHolder = e.getValue();
+ if (jobConsumerHolder.accept(initialState.getJobType())) {
+ jobConsumerHolder.consumer.execute(initialState, listener, callback);
+ return;
+ }
+ }
+ throw new IllegalArgumentException("No JobConsumer able to process a job of type "+initialState.getJobType()+" can be found in this instance.");
+ }
+
+
+ /**
+ * Holds job consumers and configures a JobTypeValve delegating to the JobConsumer implementation if it implements that interface.
+ */
+ private class JobConsumerHolder implements JobTypeValve, Closeable {
+ private final JobConsumer consumer;
+ private final Set<Types.JobType> jobTypes;
+
+ public JobConsumerHolder(JobConsumer consumer, Map<Object, Object> properties) {
+ this.consumer = consumer;
+ if ( consumer instanceof JobTypeValve) {
+ jobTypes = ImmutableSet.of();
+ } else {
+ jobTypes = Types.jobType(PropertiesUtil.toStringArray(properties.get(JobConsumer.JOB_TYPES)));
+ }
+ }
+
+ @Override
+ public boolean accept(@Nonnull Types.JobType jobType) {
+ if ( consumer instanceof JobTypeValve) {
+ return ((JobTypeValve) consumer).accept(jobType);
+ }
+ return jobTypes.contains(jobType);
+ }
+
+ public void close() {
+ // nothing to do at the moment.
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java
new file mode 100644
index 0000000..028dc3b
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java
@@ -0,0 +1,107 @@
+/*
+ * 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.jobs.impl;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobUpdate;
+import org.apache.sling.jobs.JobUpdateBuilder;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+
+/**
+ */
+public class JobUpdateBuilderImpl implements JobUpdateBuilder {
+
+ private final String jobId;
+ private final Job job;
+ private JobUpdate.JobUpdateCommand command;
+ private final ImmutableMap.Builder<String, Object> updateProperties = ImmutableMap.builder();
+
+ /**
+ * Create a JobUpdateBuilder from a job.
+ * @param job
+ */
+ public JobUpdateBuilderImpl(@Nonnull Job job) {
+ this.job = job;
+ this.jobId = null;
+ }
+
+ public JobUpdateBuilderImpl(@Nonnull String jobId) {
+ this.jobId = jobId;
+ this.job = null;
+
+ }
+
+ /**
+ * Set the JobUpdateCommand
+ * @param command the command.
+ * @return this JobBuilder instance.
+ */
+ @Nonnull
+ @Override
+ public JobUpdateBuilder command(@Nonnull JobUpdate.JobUpdateCommand command) {
+ this.command = command;
+ return this;
+ }
+
+ /**
+ * Set a property to update.
+ * @param name the name of the property
+ * @param value the value of the property which may be null. To remove the property set the value to JobUpdate.JobPropertyAction.REMOVE.
+ * @return this JobBuilder instance.
+ */
+ @Nonnull
+ @Override
+ public JobUpdateBuilder put(@Nonnull String name, @Nullable Object value) {
+ if ( value == null) {
+ this.updateProperties.put(name, JobUpdate.JobPropertyAction.REMOVE);
+ } else {
+ this.updateProperties.put(name, value);
+ }
+ return this;
+ }
+
+ @Nonnull
+ @Override
+ public JobUpdateBuilder putAll(@Nonnull Map<String, Object> properties) {
+ this.updateProperties.putAll(properties);
+ return this;
+ }
+
+
+ /**
+ * Build the JobUpdate.
+ * @return the JobUpdate.
+ */
+ @Nonnull
+ @Override
+ public JobUpdate build() {
+ if ( job != null) {
+ return new JobUpdateImpl(job, command, updateProperties.build());
+ } else if ( command == JobUpdate.JobUpdateCommand.ABORT_JOB || command == JobUpdate.JobUpdateCommand.STOP_JOB) {
+ return new JobUpdateImpl(jobId, command);
+ } else {
+ throw new IllegalStateException("Only possible to abort or stop a job by ID alone ");
+ }
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobUpdateImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobUpdateImpl.java
new file mode 100644
index 0000000..03f20cf
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/JobUpdateImpl.java
@@ -0,0 +1,234 @@
+/*
+ * 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.jobs.impl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobUpdate;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.jobs.impl.spi.MapValueAdapter;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Represents messages sent to the Job via a message queue.
+ * Abort, stop and update messages should be sent via a priority queue.
+ * Start messages should be sent by a processing queue.
+ */
+public class JobUpdateImpl implements MapValueAdapter, JobUpdate {
+ private static final long TTL = 1000 * 60;
+ private long updateTimestamp;
+ private long expires;
+ private JobUpdateCommand command;
+ private Types.JobQueue jobQueue;
+ private String id;
+ private Job.JobState jobState;
+ private Map<String, Object> properties;
+ private int retryCount;
+ private int numberOfRetries;
+ private long startedAt;
+ private long createdAt;
+ private long finishedAt;
+ private String resultMessage;
+ private Types.JobType jobType;
+
+
+ /**
+ * Create an update message using a job, command and update properties. Only the update properties will in the update.
+ * The job will be used to specify the job jobQueue, job ID and job state of the update message.
+ * @param job the job
+ * @param command the command
+ * @param properties properties in the update message.
+ */
+ public JobUpdateImpl(@Nonnull Job job, @Nonnull JobUpdateCommand command, @Nonnull Map<String, Object> properties) {
+ Preconditions.checkNotNull(job, "Job argument cant be null");
+ Preconditions.checkNotNull(command, "JobCommand argument cant be null");
+ Preconditions.checkNotNull(properties, "Map of properties cant be null");
+
+ jobQueue = job.getQueue();
+ jobType = job.getJobType();
+ id = job.getId();
+ startedAt = job.getStarted();
+ createdAt = job.getCreated();
+ finishedAt = job.getFinished();
+ retryCount = job.getRetryCount();
+ jobState = job.getJobState();
+ resultMessage = job.getResultMessage();
+ numberOfRetries = job.getNumberOfRetries();
+ updateTimestamp = System.currentTimeMillis();
+ expires = updateTimestamp + TTL;
+ this.command = command;
+ this.properties = properties;
+ }
+
+ /**
+ * Create a JobUpdateImpl based on a inbound message in the form of a Map.
+ * @param message a inbound message in map form.
+ */
+ public JobUpdateImpl(@Nonnull Map<String, Object> message) {
+ Preconditions.checkNotNull(message, "Message cant be null");
+ fromMapValue(message);
+ }
+
+ public JobUpdateImpl(@Nonnull String jobId, @Nonnull JobUpdateCommand command) {
+ Preconditions.checkNotNull(jobId, "JobId argument cant be null");
+ Preconditions.checkNotNull(command, "JobUpdateCommand argument cant be null");
+ jobQueue = Types.ANY_JOB_QUEUE;
+ id = jobId;
+ updateTimestamp = System.currentTimeMillis();
+ expires = updateTimestamp + TTL;
+ jobState = Job.JobState.ANY_STATE;
+ this.command = command;
+ this.properties = ImmutableMap.of();
+
+ }
+
+
+ @Override
+ public long updateTimestamp() {
+ return updateTimestamp;
+ }
+
+ @Override
+ public long expires() {
+ return expires;
+ }
+
+ @Nonnull
+ @Override
+ public Types.JobType getJobType() {
+ return jobType;
+ }
+
+ @Nonnull
+ @Override
+ public JobUpdateCommand getCommand() {
+ return command;
+ }
+
+ @Nonnull
+ @Override
+ public Types.JobQueue getQueue() {
+ return jobQueue;
+ }
+
+ @Nonnull
+ @Override
+ public String getId() {
+ return id;
+ }
+
+ @Nonnull
+ @Override
+ public Job.JobState getState() {
+ return jobState;
+ }
+
+ @Nonnull
+ @Override
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+
+ @Override
+ public int getRetryCount() {
+ return retryCount;
+ }
+
+ @Override
+ public int getNumberOfRetries() {
+ return numberOfRetries;
+ }
+
+ @Override
+ public long getStarted() {
+ return startedAt;
+ }
+
+ @Override
+ public long getCreated() {
+ return createdAt;
+ }
+
+ @Override
+ public long getFinished() {
+ return finishedAt;
+ }
+
+ @Override
+ public String getResultMessage() {
+ return resultMessage;
+ }
+
+
+ @Override
+ public void fromMapValue(@Nullable Object mapValue) {
+ if (mapValue != null && mapValue instanceof Map) {
+ @SuppressWarnings("unchecked") Map<String, Object> m = (Map<String, Object>) mapValue;
+ jobQueue = Types.jobQueue((String) Utils.getRequired(m, "tp"));
+ jobType = Types.jobType((String)Utils.getRequired(m, "jt"));
+ id = Utils.getRequired(m, "id");
+ command = JobUpdateCommand.valueOf((String) Utils.getRequired(m, "cm"));
+ updateTimestamp = Utils.getRequired(m, "ts");
+ expires = Utils.getRequired(m, "ex");
+ if (command == JobUpdateCommand.UPDATE_JOB || command == JobUpdateCommand.START_JOB || command == JobUpdateCommand.RETRY_JOB ) {
+ startedAt = Utils.getOptional(m, "startedAt", 0L);
+ createdAt = Utils.getOptional(m, "createdAt", 0L);
+ finishedAt = Utils.getOptional(m, "finishedAt", 0L);
+ retryCount = Utils.getOptional(m, "retryCount", 0);
+ numberOfRetries = Utils.getOptional(m, "nRetries", 10);
+ jobState = Job.JobState.valueOf(Utils.getOptional(m, "jobState", Job.JobState.QUEUED.toString()));
+ resultMessage = Utils.getOptional(m, "resultMessage", null);
+ properties = Utils.getOptional(m, "properties", new HashMap<String, Object>());
+ } else {
+ properties = new HashMap<String, Object>();
+ }
+ } else {
+ throw new IllegalArgumentException("Cant populate JobImpl from "+mapValue);
+ }
+ }
+
+ @Override
+ @Nonnull
+ public Object toMapValue() {
+ ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
+ builder.put("tp", jobQueue.toString());
+ builder.put("jt",jobType.toString());
+ builder.put("id",id);
+ builder.put("cm", command.toString());
+ builder.put("ts", this.updateTimestamp);
+ builder.put("ex", expires);
+ if ( command == JobUpdateCommand.UPDATE_JOB || command == JobUpdateCommand.START_JOB || command == JobUpdateCommand.RETRY_JOB ) {
+ builder.put("retryCount", retryCount);
+ builder.put("nRetries", numberOfRetries);
+ builder.put("startedAt", startedAt);
+ builder.put("createdAt", createdAt);
+ builder.put("finishedAt", finishedAt);
+ builder.put("jobState", jobState.toString());
+ builder.put("resultMessage", resultMessage);
+ builder.put("properties", ImmutableMap.builder().putAll(properties).build());
+
+ }
+ return builder.build();
+ }
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/ManagerSubscriber.java b/src/main/java/org/apache/sling/jobs/impl/ManagerSubscriber.java
new file mode 100644
index 0000000..76ff8e2
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/ManagerSubscriber.java
@@ -0,0 +1,51 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.felix.scr.annotations.*;
+import org.apache.sling.jobs.JobManager;
+import org.apache.sling.jobs.JobUpdateListener;
+import org.apache.sling.mom.Subscriber;
+import org.apache.sling.mom.Types;
+
+import java.util.Map;
+
+/**
+ * Listens to a topic to retrieve control messages.
+ */
+@Component(immediate = true, metatype = true)
+@Service(value = Subscriber.class)
+@Properties({
+ @Property(name= Subscriber.TOPIC_NAMES_PROP, cardinality = Integer.MAX_VALUE, value = {"sling/jobupdates"} )
+})
+public class ManagerSubscriber implements Subscriber {
+
+
+ @Reference
+ private JobManager jobManager;
+
+
+ @Override
+ public void onMessage(Types.TopicName topic, Map<String, Object> message) {
+ if (jobManager instanceof JobUpdateListener) {
+ ((JobUpdateListener) jobManager).update(new JobUpdateImpl(message));
+ }
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/OutboundJobUpdateListener.java b/src/main/java/org/apache/sling/jobs/impl/OutboundJobUpdateListener.java
new file mode 100644
index 0000000..d6bebad
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/OutboundJobUpdateListener.java
@@ -0,0 +1,65 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.JobUpdate;
+import org.apache.sling.jobs.JobUpdateListener;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.mom.QueueManager;
+import org.apache.sling.mom.TopicManager;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Sends messages out to JMS Queues or topics. Normally called by the local JobManager Implementation.
+ * Uses a TopicManager or QueueManager to perform the send operation.
+ */
+public class OutboundJobUpdateListener implements JobUpdateListener {
+
+
+ private boolean active;
+ private final TopicManager topicManager;
+ private final QueueManager queueManager;
+
+ public OutboundJobUpdateListener(TopicManager topicManager, QueueManager queueManager ) {
+ this.topicManager = topicManager;
+ this.queueManager = queueManager;
+ active = true;
+ }
+
+ public void dispose() {
+ active = false;
+ }
+
+
+ @Override
+ public void update(@Nonnull JobUpdate update) {
+ if ( active ) {
+ switch(update.getCommand()) {
+ case START_JOB:
+ queueManager.add(update.getQueue().asQueueName(), Utils.toMapValue(update));
+ break;
+ default:
+ topicManager.publish(update.getQueue().asTopicName(), update.getCommand().asCommandName(), Utils.toMapValue(update));
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/Utils.java b/src/main/java/org/apache/sling/jobs/impl/Utils.java
new file mode 100644
index 0000000..53d4475
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/Utils.java
@@ -0,0 +1,150 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.impl.spi.MapValueAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.io.File;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.nio.charset.Charset;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ */
+public class Utils {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
+ private static final Charset UTF8 = Charset.forName("UTF-8") ;
+ private static final String PROCESS_NAME = generateUniqueNamespace();
+ // Set the counter to the class load time.
+ private static final AtomicLong idCounter = new AtomicLong(System.currentTimeMillis());
+
+ /**
+ * Gets a string
+ * @return
+ */
+ @Nonnull
+ private static String generateUniqueNamespace() {
+ String macAddress = null;
+ // get the MAC address of the primary interface, failing that use a fake.
+ try {
+ for ( Enumeration<NetworkInterface> netInterfaceE = NetworkInterface.getNetworkInterfaces(); netInterfaceE.hasMoreElements();) {
+ NetworkInterface netInterface = netInterfaceE.nextElement();
+ byte[] hw = netInterface.getHardwareAddress();
+ if ( !netInterface.isLoopback() && !netInterface.isVirtual() && hw != null) {
+ macAddress = tohex(hw);
+ LOGGER.info("Job IDs seeded with MAC Address from interface {} ", netInterface);
+ break;
+ }
+ }
+ if ( macAddress == null) {
+ LOGGER.info("No MAC address available, seeding JobID from startup time.");
+ macAddress = "fake-" + System.currentTimeMillis();
+ }
+ } catch (SocketException e) {
+ LOGGER.warn("Unable to get MAC address, defaulting to fake ", e);
+ }
+ long processID;
+ try {
+ // most JVMs.
+ processID = Long.parseLong(java.lang.management.ManagementFactory.getRuntimeMXBean().getName().split("@")[0]);
+ } catch (Exception e) {
+ try {
+ // most Linux kernels.
+ processID = Long.parseLong(new File("/proc/self").getCanonicalFile().getName());
+ } catch (Exception e1) {
+ LOGGER.warn("Unable to get ProcessID by address, defaulting to fake ", e);
+ processID = System.currentTimeMillis(); // this will be way beyond any process ID.
+ }
+ }
+ String baseId = macAddress + "/" + processID+ "/";
+ LOGGER.info("Job IDS base is {} ", baseId);
+ return baseId;
+ }
+
+ @Nonnull
+ private static String tohex(@Nonnull byte[] bytes) {
+ StringBuilder sb = new StringBuilder();
+ for( byte b : bytes) {
+ sb.append(String.format("%02x",b));
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Generate an ID based on the unique name of the jvm process and a counter.
+ * @return
+ */
+ @Nonnull
+ public static String generateId() {
+ try {
+ return Utils.tohex(MessageDigest.getInstance("SHA1").digest((Utils.PROCESS_NAME+idCounter.incrementAndGet()).getBytes(UTF8)));
+ } catch (NoSuchAlgorithmException nsae) {
+ throw new RuntimeException("SHA1 not supported", nsae);
+ }
+ }
+
+ @Nonnull
+ public static Map<String, Object> toMapValue(@Nonnull Object msg) {
+ if (msg instanceof Map) {
+ //noinspection unchecked
+ return (Map<String, Object>) msg;
+ } else if (msg instanceof MapValueAdapter) {
+ return toMapValue(((MapValueAdapter) msg).toMapValue());
+ }
+ throw new IllegalArgumentException("Unable to convert "+msg.getClass()+" to a Map.");
+ }
+
+
+ @Nonnull
+ public static <T> T getRequired(@Nonnull Map<String, Object> m, @Nonnull String name) {
+ if (m.containsKey(name)) {
+ //noinspection unchecked
+ if ( m.get(name) != null) {
+ return (T) m.get(name);
+ }
+ }
+ throw new IllegalArgumentException("Required key "+name+" is missing from "+m);
+ }
+
+ @Nullable
+ public static <T> T getOptional(@Nonnull Map<String, Object> m, @Nonnull String name, @Nullable T defaultValue) {
+ if (m.containsKey(name)) {
+ //noinspection unchecked
+ Object o = m.get(name);
+ if ( defaultValue instanceof Integer && o instanceof Long) {
+ return (T)(Integer) ((Long) o).intValue();
+ } else if ( defaultValue instanceof Float && o instanceof Double) {
+ return (T)(Float) ((Double) o).floatValue();
+ }
+ return (T) o;
+ }
+ return defaultValue;
+ }
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java b/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java
new file mode 100644
index 0000000..f9ccd76
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java
@@ -0,0 +1,28 @@
+/*
+ * 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.jobs.impl.spi;
+
+import org.apache.sling.jobs.Job;
+
+/**
+ * Starts a job, used by the JobBuilderImpl to perform the add operation.
+ */
+public interface JobStarter {
+ Job start(Job j);
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
new file mode 100644
index 0000000..e33968f
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
@@ -0,0 +1,70 @@
+/*
+ * 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.jobs.impl.spi;
+
+import org.apache.sling.jobs.Job;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+/**
+ * Provides JobStorage local to the JVM. Implementation may or may not decide to persist over restarts, page, etc
+ */
+public interface JobStorage {
+
+
+ /**
+ * Get a Job by ID.
+ * @param jobId the job ID to get.
+ * @return the job or null of the job doesn't exist.
+ */
+ @Nullable
+ Job get(@Nonnull String jobId);
+
+ /**
+ * Put a Job into the Job Storage.
+ * @param job the job.
+ * @return the job just added.
+ */
+ @Nonnull
+ Job put(@Nonnull Job job);
+
+ /**
+ * Remove the Job
+ * @param jobId
+ * @return the job removed or null if not present.
+ */
+ @Nullable
+ Job remove(@Nonnull String jobId);
+
+ /**
+ * Remove the Job, returning the job removed.
+ * @param job the job to remove.
+ * @return the job removed, if the the job was present, otherwise null.
+ */
+ @Nullable
+ Job remove(@Nonnull Job job);
+
+ /**
+ * Dispose of the JobStorage.
+ */
+ void dispose();
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
new file mode 100644
index 0000000..2266ee6
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.jobs.impl.spi;
+
+/**
+ * Objects that can be converted to and from a Map are expected to extend this base class.
+ * The values contained in the Map are expected to be Maps or values that can be serialised into most
+ * common formats. It would be safe to use json or yaml as an example of a common format.
+ *
+ */
+public interface MapValueAdapter {
+
+
+ /**
+ * Populate the object from a map value.
+ * @param mapValue
+ */
+ void fromMapValue(Object mapValue);
+
+ /**
+ * Adapt the object into a value suitable for use in a map to be serialised by standard map -> json,yaml writers.
+ * @return a value, which may be a primitive, an array or a map of primitives.
+ */
+ Object toMapValue();
+
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorage.java b/src/main/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorage.java
new file mode 100644
index 0000000..222e339
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorage.java
@@ -0,0 +1,79 @@
+/*
+ * 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.jobs.impl.storage;
+
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * An unbounded local JVM job store.
+ */
+public class InMemoryJobStorage implements JobStorage {
+
+
+ private Map<String, Job> store = new ConcurrentHashMap<String, Job>();
+
+ @Nullable
+ @Override
+ public Job get(@Nonnull String jobId) {
+ check();
+ return store.get(jobId);
+ }
+
+ @Nonnull
+ @Override
+ public Job put(@Nonnull Job job) {
+ check();
+ store.put(job.getId(), job);
+ return job;
+ }
+
+ @Nullable
+ @Override
+ public Job remove(@Nonnull String jobId) {
+ check();
+ Job j = store.get(jobId);
+ store.remove(jobId);
+ return j;
+ }
+
+ @Nullable
+ @Override
+ public Job remove(@Nonnull Job job) {
+ check();
+ return remove(job.getId());
+ }
+
+ private void check() {
+ if ( store == null) {
+ throw new IllegalStateException("Job store already closed.");
+ }
+ }
+
+ @Override
+ public void dispose() {
+ store.clear();
+ store = null;
+ }
+}
diff --git a/src/main/java/org/apache/sling/jobs/package-info.java b/src/main/java/org/apache/sling/jobs/package-info.java
new file mode 100644
index 0000000..3a61230
--- /dev/null
+++ b/src/main/java/org/apache/sling/jobs/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+@Version("1.0.0")
+package org.apache.sling.jobs;
+
+import aQute.bnd.annotation.Version;
+
diff --git a/src/test/java/org/apache/sling/jobs/impl/JobBuilderImplTest.java b/src/test/java/org/apache/sling/jobs/impl/JobBuilderImplTest.java
new file mode 100644
index 0000000..2da7293
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/JobBuilderImplTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobController;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.jobs.impl.spi.JobStarter;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests job builder.
+ */
+public class JobBuilderImplTest {
+
+ @Mock
+ private JobStarter jobStarter;
+
+ private Queue<Job> queue;
+ @Mock
+ private JobController jobController;
+
+ public JobBuilderImplTest() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Before
+ public void before() {
+ queue = new ArrayBlockingQueue<Job>(1);
+ Mockito.when(jobStarter.start(Mockito.any(Job.class))).then(new Answer<Job>() {
+ @Override
+ public Job answer(InvocationOnMock invocationOnMock) throws Throwable {
+ queue.add((Job) invocationOnMock.getArguments()[0]);
+ return (Job) invocationOnMock.getArguments()[0];
+ }
+ });
+ }
+
+ @Test
+ public void testAddJob() {
+ long start = System.currentTimeMillis();
+ Map<String, Object> testMap = new HashMap<String, Object>();
+ testMap.put("job.name", "Jobname");
+ Job queuedJob = new JobBuilderImpl(jobStarter, Types.jobQueue("testtopic"), Types.jobType("testtype")).addProperties(testMap).add();
+ assertEquals(1, queue.size());
+ Job fromQueue = queue.remove();
+ assertEquals(queuedJob, fromQueue);
+ assertEquals(Types.jobQueue("testtopic"), fromQueue.getQueue());
+ assertEquals("Jobname", fromQueue.getProperties().get("job.name"));
+ assertNotNull(fromQueue.getId());
+ long now = System.currentTimeMillis();
+ assertTrue(fromQueue.getCreated() >= start);
+ assertTrue(fromQueue.getCreated() <= now);
+ assertEquals(Job.JobState.CREATED, fromQueue.getJobState());
+ assertNull(fromQueue.getController());
+ fromQueue.setJobController(jobController);
+ assertEquals(jobController, fromQueue.getController());
+ fromQueue.removeJobController();
+ assertNull(fromQueue.getController());
+ assertEquals("", fromQueue.getResultMessage());
+ assertEquals(0, fromQueue.getFinished());
+ assertEquals(0, fromQueue.getStarted());
+ assertEquals(0, fromQueue.getNumberOfRetries());
+ assertEquals(0, fromQueue.getRetryCount());
+
+ }
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/JobImplTest.java b/src/test/java/org/apache/sling/jobs/impl/JobImplTest.java
new file mode 100644
index 0000000..fa3e845
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/JobImplTest.java
@@ -0,0 +1,387 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.*;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+/**
+ */
+public class JobImplTest {
+
+ private Job job;
+ private String jobId;
+ @Mock
+ private JobController jobController;
+ private final long before = System.currentTimeMillis();
+ private long after = System.currentTimeMillis();
+ private boolean jobAborted;
+ private boolean jobStopped;
+
+ public JobImplTest() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ jobAborted = true;
+ return null;
+ }
+ }).when(jobController).abort();
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ jobStopped = true;
+ return null;
+ }
+ }).when(jobController).stop();
+
+ Thread.sleep(1);
+ Map<String, Object> properties = new HashMap<String, Object>();
+ properties.put("job.name", "Jobname");
+ jobId = Utils.generateId();
+ job = new JobImpl(Types.jobQueue("testtopic"), jobId, Types.jobType("testtype"), properties);
+ Thread.sleep(1);
+ after = System.currentTimeMillis();
+
+
+
+ }
+
+
+ @Test
+ public void testGetTopic() throws Exception {
+ assertEquals(Types.jobQueue("testtopic"), job.getQueue());
+ }
+
+ @Test
+ public void testGetId() throws Exception {
+ assertEquals(jobId, job.getId());
+ }
+
+ @Test
+ public void testGetProperties() throws Exception {
+ assertEquals(1, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ }
+
+ @Test
+ public void testGetRetryCount() throws Exception {
+ assertEquals(0,job.getRetryCount());
+ }
+
+ @Test
+ public void testGetNumberOfRetries() throws Exception {
+ assertEquals(0,job.getNumberOfRetries());
+
+ }
+
+ @Test
+ public void testGetStarted() throws Exception {
+ assertEquals(0, job.getStarted());
+
+ }
+
+ @Test
+ public void testGetCreated() throws Exception {
+ long created = job.getCreated();
+ assertTrue(created > before && created < after);
+
+ }
+
+ @Test
+ public void testGetJobState() throws Exception {
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+
+ }
+
+ @Test
+ public void testSetState() throws Exception {
+ assertEquals(Job.JobState.CREATED,job.getJobState());
+ job.setState(Job.JobState.STOPPED);
+ assertEquals(Job.JobState.STOPPED, job.getJobState());
+
+ }
+
+ @Test
+ public void testGetFinished() throws Exception {
+ assertEquals(0,job.getFinished());
+
+ }
+
+ @Test
+ public void testGetResultMessage() throws Exception {
+ assertEquals("",job.getResultMessage());
+
+ }
+
+ @Test
+ public void testGetController() throws Exception {
+ assertEquals(null,job.getController());
+
+ }
+
+ @Test
+ public void testSetJobController() throws Exception {
+ assertEquals(null,job.getController());
+ job.setJobController(jobController);
+ assertEquals(jobController,job.getController());
+
+ }
+
+ @Test
+ public void testRemoveJobController() throws Exception {
+ assertEquals(null,job.getController());
+ job.setJobController(jobController);
+ assertEquals(jobController, job.getController());
+ job.removeJobController();
+ assertEquals(null, job.getController());
+
+ }
+
+ @Test
+ public void testUpdate() throws Exception {
+ JobImpl jobImpl = (JobImpl) job;
+ JobUpdateBuilder jobUpdateBuilder = jobImpl.newJobUpdateBuilder();
+ assertNotNull(jobUpdateBuilder);
+ Map<String,Object> updateProperties = new HashMap<String, Object>();
+ updateProperties.put("updateprop", "updatevalue");
+ JobUpdate jobUpdate = jobUpdateBuilder
+ .command(JobUpdate.JobUpdateCommand.START_JOB)
+ .putAll(updateProperties)
+ .put("extra", "extravalue")
+ .build();
+
+ jobImpl.update(jobUpdate);
+
+ // Job update does not change the job state even if the command says so.
+ // the state change comes from a job starter.
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(3, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("extravalue", job.getProperties().get("extra"));
+ assertEquals("updatevalue", job.getProperties().get("updateprop"));
+
+ JobImpl jobImp = new JobImpl(jobUpdate);
+ assertEquals(job.getId(), jobImp.getId());
+
+
+ // update with some, but not all the new properties.
+ updateProperties = new HashMap<String, Object>();
+ updateProperties.put("updateprop", "updatevalue2");
+
+
+ jobImpl.update(jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.UPDATE_JOB)
+ .putAll(updateProperties)
+ .put("extra", "extravalue")
+ .build());
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(3, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("extravalue", job.getProperties().get("extra"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+
+ // remove a property.
+ jobImpl.update(jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.UPDATE_JOB)
+ .put("extra", JobUpdate.JobPropertyAction.REMOVE)
+ .build());
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(2, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+
+ // remove a property.
+ JobUpdate retryUpdate = jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.RETRY_JOB)
+ .put("updateprop", "updatevalue2")
+ .build();
+ set(retryUpdate, "numberOfRetries", 10);
+ set(retryUpdate, "retryCount", 10);
+ jobImpl.update(retryUpdate);
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(2, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+ assertEquals(20, job.getNumberOfRetries());
+ assertEquals(10,job.getRetryCount());
+
+
+
+ // Abort the job without a controller present.
+ jobAborted = false;
+ jobStopped = false;
+ jobImpl.update(jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.ABORT_JOB)
+ .put("extra2", "newextra2")
+ .build());
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(2, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+ assertFalse(jobAborted);
+ assertFalse(jobStopped);
+
+ // set a controller as if the job was stated.
+ job.setJobController(jobController);
+ jobAborted = false;
+ jobStopped = false;
+
+ // abort the job
+ jobImpl.update(jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.ABORT_JOB)
+ .put("extra2", "newextra3")
+ .build());
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ // abort does not update properties.
+ assertEquals(2, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+ assertTrue(jobAborted);
+ assertFalse(jobStopped);
+
+ jobAborted = false;
+ jobStopped = false;
+
+ JobUpdate wrongOrder = jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.UPDATE_JOB)
+ .put("extra2", "newextra4")
+ .build();
+
+ Thread.sleep(2);
+ // abort the job
+ jobImpl.update(jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.STOP_JOB)
+ .put("extra2", "newextra4")
+ .build());
+
+ assertEquals(Job.JobState.CREATED, job.getJobState());
+ assertEquals(2, job.getProperties().size());
+ assertEquals("Jobname", job.getProperties().get("job.name"));
+ assertEquals("updatevalue2", job.getProperties().get("updateprop"));
+ assertFalse(jobAborted);
+ assertTrue(jobStopped);
+
+
+ try {
+ jobImpl.update(wrongOrder);
+ fail("Update should have been rejected due to out of sequence");
+ } catch ( IllegalStateException e) {
+ // ok
+ }
+
+ JobUpdate wrongId = new JobUpdateImpl(Utils.generateId(), JobUpdate.JobUpdateCommand.UPDATE_JOB);
+ try {
+ jobImpl.update(wrongId);
+ fail("Update should have been rejected due wrong Id");
+ } catch ( IllegalArgumentException e) {
+ // ok
+ }
+
+ JobUpdate expired = jobImpl.newJobUpdateBuilder()
+ .command(JobUpdate.JobUpdateCommand.STOP_JOB)
+ .put("extra2", "newextra4")
+ .build();
+ set(expired, "expires", System.currentTimeMillis());
+ Thread.sleep(1);
+ try {
+ jobImpl.update(expired);
+ fail("Update should have expired");
+ } catch ( IllegalStateException e) {
+ // ok
+ }
+
+ try {
+ JobUpdate failJobUpdate = new JobUpdateBuilderImpl(job.getId()).command(JobUpdate.JobUpdateCommand.START_JOB).build();
+ fail("should not be able to build a job by ID only");
+ } catch ( IllegalStateException e) {
+ // ok
+ JobUpdate stopJob = new JobUpdateBuilderImpl(job.getId()).command(JobUpdate.JobUpdateCommand.STOP_JOB).build();
+ JobUpdate abortJob = new JobUpdateBuilderImpl(job.getId()).command(JobUpdate.JobUpdateCommand.ABORT_JOB).build();
+ }
+
+
+ }
+
+ private void set(Object obj, String name, Object v) throws NoSuchFieldException, IllegalAccessException {
+ Field f = obj.getClass().getDeclaredField(name);
+ f.setAccessible(true);
+ f.set(obj, v);
+ }
+
+
+ @Test
+ public void testNewJobUpdateBuilder() throws Exception {
+ long updateBefore = System.currentTimeMillis();
+ Thread.sleep(1);
+
+ JobImpl jobImpl = (JobImpl) job;
+ JobUpdateBuilder jobUpdateBuilder = jobImpl.newJobUpdateBuilder();
+ assertNotNull(jobUpdateBuilder);
+ Map<String,Object> updateProperties = new HashMap<String, Object>();
+ updateProperties.put("updateprop", "updatevalue");
+ JobUpdate jobUpdate = jobUpdateBuilder
+ .command(JobUpdate.JobUpdateCommand.START_JOB)
+ .putAll(updateProperties)
+ .put("extra", "extravalue")
+ .build();
+
+
+ Thread.sleep(1);
+ long updateAfter = System.currentTimeMillis();
+ assertNotNull(jobUpdate);
+ assertEquals(jobId, jobUpdate.getId());
+ assertEquals(Types.jobQueue("testtopic"), jobUpdate.getQueue());
+ assertEquals("", jobUpdate.getResultMessage());
+ assertTrue(jobUpdate.expires() > updateAfter);
+ assertTrue(jobUpdate.updateTimestamp() > updateBefore && jobUpdate.updateTimestamp() < updateAfter);
+ assertTrue(jobUpdate.getCreated() > before && jobUpdate.getCreated() < after);
+ assertEquals(JobUpdate.JobUpdateCommand.START_JOB, jobUpdate.getCommand());
+ assertEquals(Job.JobState.CREATED, jobUpdate.getState());
+ assertEquals(0, jobUpdate.getStarted());
+ assertEquals(0,jobUpdate.getFinished());
+ assertEquals(0,jobUpdate.getRetryCount());
+ assertEquals(0,jobUpdate.getNumberOfRetries());
+ assertEquals(2,jobUpdate.getProperties().size());
+ assertEquals("extravalue", jobUpdate.getProperties().get("extra"));
+ assertEquals("updatevalue",jobUpdate.getProperties().get("updateprop"));
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/JobManagerImplTest.java b/src/test/java/org/apache/sling/jobs/impl/JobManagerImplTest.java
new file mode 100644
index 0000000..5cab281
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/JobManagerImplTest.java
@@ -0,0 +1,264 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.*;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+import org.apache.sling.jobs.impl.storage.InMemoryJobStorage;
+import org.apache.sling.mom.*;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+
+import static org.junit.Assert.*;
+
+/**
+ */
+public class JobManagerImplTest {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobManagerImplTest.class);
+ private JobManager jobManager;
+ private JobStorage jobStorage;
+ private OutboundJobUpdateListener messageSender;
+ @Mock
+ private TopicManager topicManager;
+ @Mock
+ private QueueManager queueManager;
+ private Map<org.apache.sling.mom.Types.TopicName, Queue<QueueEntry>> topicQueues;
+ private Map<org.apache.sling.mom.Types.QueueName, Queue<QueueEntry>> messageQueues;
+
+ public JobManagerImplTest() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ topicQueues = new HashMap<org.apache.sling.mom.Types.TopicName, Queue<QueueEntry>>();
+ messageQueues = new HashMap<org.apache.sling.mom.Types.QueueName, Queue<QueueEntry>>();
+ //noinspection unchecked
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ org.apache.sling.mom.Types.TopicName topic = (org.apache.sling.mom.Types.TopicName) invocationOnMock.getArguments()[0];
+ org.apache.sling.mom.Types.CommandName command = (org.apache.sling.mom.Types.CommandName) invocationOnMock.getArguments()[1];
+ @SuppressWarnings("unchecked") Map<String, Object> properties = (Map<String, Object>) invocationOnMock.getArguments()[2];
+ LOGGER.info("Topic Manager publish {} {} {} ", new Object[]{ topic, command, properties });
+ Queue<QueueEntry> queue = topicQueues.get(topic);
+ if ( queue == null) {
+ queue = new ArrayBlockingQueue<QueueEntry>(100);
+ topicQueues.put(topic, queue);
+
+ }
+ queue.add(new QueueEntry(command, properties));
+ return null;
+ }
+ }).when(topicManager)
+ .publish(Mockito.any(org.apache.sling.mom.Types.TopicName.class), Mockito.any(org.apache.sling.mom.Types.CommandName.class), Mockito.any(Map.class));
+
+ //noinspection unchecked
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ org.apache.sling.mom.Types.QueueName topic = (org.apache.sling.mom.Types.QueueName) invocationOnMock.getArguments()[0];
+ @SuppressWarnings("unchecked") Map<String, Object> properties = (Map<String, Object>) invocationOnMock.getArguments()[1];
+ LOGGER.info("Queue Manager add {} {} {} ", new Object[]{ topic, properties });
+ Queue<QueueEntry> queue = messageQueues.get(topic);
+ if ( queue == null) {
+ queue = new ArrayBlockingQueue<QueueEntry>(100);
+ messageQueues.put(topic, queue);
+
+ }
+ queue.add(new QueueEntry(properties));
+ return null;
+ }
+ }).when(queueManager)
+ .add(Mockito.any(org.apache.sling.mom.Types.QueueName.class), Mockito.any(Map.class));
+
+ messageSender = new OutboundJobUpdateListener(topicManager, queueManager);
+ jobStorage = new InMemoryJobStorage();
+ jobManager = new JobManagerImpl(jobStorage, messageSender);
+
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ messageSender.dispose();
+ jobStorage.dispose();
+ }
+
+ @Test
+ public void testCreateJob() throws Exception {
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ Job job = createJob(testId);
+
+ assertEquals(testId, job.getProperties().get("testid"));
+ org.apache.sling.mom.Types.TopicName topicName = org.apache.sling.mom.Types.topicName("testtopic");
+ assertNotNull(messageQueues.get(topicName));
+ assertEquals(1, messageQueues.get(topicName).size());
+ QueueEntry qe = messageQueues.get(topicName).remove();
+ assertNotNull(qe);
+ Map<String, Object> jobProperties = qe.getProperties();
+
+ // pretend to consume the job message
+ JobImpl dequeuedJob = new JobImpl(new JobUpdateImpl(jobProperties));
+ assertEquals(job.getId(), dequeuedJob.getId());
+ // the operation of the job object is tested in other unit tests, don't repeat that here.
+ assertEquals(testId, dequeuedJob.getProperties().get("testid"));
+
+ }
+
+ @Test
+ public void testGetJobById() throws Exception {
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ Job job = createJob(testId);
+
+
+ Job searchedJob = jobManager.getJobById(job.getId());
+ assertNotNull(searchedJob);
+ assertEquals(job.getId(), searchedJob.getId());
+ assertEquals(testId, searchedJob.getProperties().get("testid"));
+
+ }
+
+ private Job createJob(String testId) {
+ Map<String, Object> testProps = new HashMap<String, Object>();
+ testProps.put("job.name", "Jobname ");
+ testProps.put("testid", testId);
+ Job job = jobManager.newJobBuilder(Types.jobQueue("testtopic"), Types.jobType("testtype"))
+ .addProperties(testProps)
+ .add();
+ assertNotNull(job);
+ return job;
+ }
+
+
+ @Test
+ public void testStopJobById() throws Exception {
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ Job job = createJob(testId);
+ Thread.sleep(10);
+ jobManager.stopJobById(job.getId());
+ Queue<QueueEntry> messageQ = messageQueues.get(org.apache.sling.mom.Types.queueName("testtopic"));
+ Queue<QueueEntry> topicQ = topicQueues.get(org.apache.sling.mom.Types.topicName("testtopic"));
+ assertEquals(1,messageQ.size());
+ QueueEntry qe = messageQ.remove();
+ assertNotNull(qe);
+ Map<String, Object> jobProperties = qe.getProperties();
+
+ // get the job off the queue.
+ JobImpl dequeuedJob = new JobImpl(new JobUpdateImpl(jobProperties));
+ assertEquals(job.getId(), dequeuedJob.getId());
+ // the operation of the job object is tested in other unit tests, don't repeat that here.
+ assertEquals(testId, dequeuedJob.getProperties().get("testid"));
+
+
+ // get any messages sent to the topic.
+ assertEquals(1, topicQ.size());
+ QueueEntry stoppedQE = topicQ.remove();
+ Map<String, Object> stoppedJobProperties = stoppedQE.getProperties();
+ assertNotNull(stoppedJobProperties);
+ JobImpl stoppedJob = new JobImpl(new JobUpdateImpl(stoppedJobProperties));
+ assertEquals(JobUpdate.JobUpdateCommand.STOP_JOB.asCommandName(), stoppedQE.getCommand());
+ assertEquals(job.getId(), stoppedJob.getId());
+ // the stop message to the topic wont have any properties.
+
+
+
+ }
+
+ @Test
+ public void testAbortJob() throws Exception {
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ Job job = createJob(testId);
+ Thread.sleep(10);
+ jobManager.abortJob(job.getId());
+ LOGGER.info("Message Queues {}", messageQueues);
+ LOGGER.info("Topic Queues {}", topicQueues);
+ Queue<QueueEntry> messageQ = messageQueues.get(org.apache.sling.mom.Types.queueName("testtopic"));
+ Queue<QueueEntry> topicQ = topicQueues.get(org.apache.sling.mom.Types.topicName("testtopic"));
+ assertEquals(1,messageQ.size());
+ QueueEntry qe = messageQ.remove();
+ assertNotNull(qe);
+ Map<String, Object> jobProperties = qe.getProperties();
+
+ // get the job off the queue.
+ JobImpl dequeuedJob = new JobImpl(new JobUpdateImpl(jobProperties));
+ assertEquals(job.getId(), dequeuedJob.getId());
+ // the operation of the job object is tested in other unit tests, dont repeat that here.
+ assertEquals(testId, dequeuedJob.getProperties().get("testid"));
+
+
+ // get any messages sent to the topic.
+ assertEquals(1, topicQ.size());
+ QueueEntry stoppedQE = topicQ.remove();
+ Map<String, Object> stoppedJobProperties = stoppedQE.getProperties();
+ assertNotNull(stoppedJobProperties);
+ JobImpl stoppedJob = new JobImpl(new JobUpdateImpl(stoppedJobProperties));
+ assertEquals(JobUpdate.JobUpdateCommand.ABORT_JOB.asCommandName(), stoppedQE.getCommand());
+ assertEquals(job.getId(), stoppedJob.getId());
+
+ }
+
+ @Test
+ public void testRetryJobById() throws Exception {
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ Job job = createJob(testId);
+ Thread.sleep(10);
+ jobManager.retryJobById(job.getId());
+ Queue<QueueEntry> messageQ = messageQueues.get(org.apache.sling.mom.Types.queueName("testtopic"));
+ Queue<QueueEntry> topicQ = topicQueues.get(org.apache.sling.mom.Types.topicName("testtopic"));
+ assertEquals(1,messageQ.size());
+ QueueEntry qe = messageQ.remove();
+ assertNotNull(qe);
+ Map<String, Object> jobProperties = qe.getProperties();
+
+ // get the job off the queue.
+ JobImpl dequeuedJob = new JobImpl(new JobUpdateImpl(jobProperties));
+ assertEquals(job.getId(), dequeuedJob.getId());
+ // the operation of the job object is tested in other unit tests, don't repeat that here.
+ assertEquals(testId, dequeuedJob.getProperties().get("testid"));
+
+
+ // get any messages sent to the topic.
+ assertEquals(1, topicQ.size());
+ QueueEntry stoppedQE = topicQ.remove();
+ Map<String, Object> stoppedJobProperties = stoppedQE.getProperties();
+ assertNotNull(stoppedJobProperties);
+ JobImpl stoppedJob = new JobImpl(new JobUpdateImpl(stoppedJobProperties));
+ assertEquals(JobUpdate.JobUpdateCommand.RETRY_JOB.asCommandName(), stoppedQE.getCommand());
+ assertEquals(job.getId(), stoppedJob.getId());
+
+ }
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/JobUpdateImplTest.java b/src/test/java/org/apache/sling/jobs/impl/JobUpdateImplTest.java
new file mode 100644
index 0000000..8b42c40
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/JobUpdateImplTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobUpdate;
+import org.apache.sling.jobs.Types;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+/**
+ */
+public class JobUpdateImplTest {
+
+ private String jobId;
+ private JobImpl job;
+
+ @Before
+ public void setUp() throws Exception {
+ Map<String, Object> properties = new HashMap<String, Object>();
+ properties.put("job.name", "Jobname");
+ jobId = Utils.generateId();
+ job = new JobImpl(Types.jobQueue("testtopic"), jobId, Types.jobType("testtype"), properties);
+ }
+
+ @Test
+ public void test() throws Exception {
+
+ Map<String, Object> properties = new HashMap<String, Object>();
+ properties.put("Update", "update value");
+ JobUpdateImpl jobUpdate = new JobUpdateImpl(job, JobUpdate.JobUpdateCommand.START_JOB, properties);
+
+ Map<String, Object> value = Utils.toMapValue(jobUpdate);
+ assertNotNull(value);
+ System.err.println("Value is " + value);
+ assertEquals(14, value.size());
+
+ JobUpdate jobUpdateCopy = new JobUpdateImpl(value);
+ assertEquals(jobUpdate.getId(), jobUpdateCopy.getId());
+ assertEquals(jobUpdate.expires(), jobUpdateCopy.expires());
+ assertEquals(jobUpdate.updateTimestamp(), jobUpdateCopy.updateTimestamp());
+ assertEquals(jobUpdate.getCreated(), jobUpdateCopy.getCreated());
+ assertEquals(jobUpdate.getStarted(), jobUpdateCopy.getStarted());
+ assertEquals(jobUpdate.getCommand(), jobUpdateCopy.getCommand());
+ assertEquals(jobUpdate.getFinished(), jobUpdateCopy.getFinished());
+ assertEquals(jobUpdate.getResultMessage(), jobUpdateCopy.getResultMessage());
+ assertEquals(jobUpdate.getNumberOfRetries(), jobUpdateCopy.getNumberOfRetries());
+ assertEquals(jobUpdate.getRetryCount(), jobUpdateCopy.getRetryCount());
+ assertEquals(jobUpdate.getState(), jobUpdateCopy.getState());
+ assertEquals(jobUpdate.getQueue(), jobUpdateCopy.getQueue());
+ Map<String, Object> originalProperties = jobUpdate.getProperties();
+ Map<String, Object> copyProperties = jobUpdateCopy.getProperties();
+ assertEquals(1, copyProperties.size());
+ assertEquals(copyProperties.size(), copyProperties.size());
+ assertEquals("update value", originalProperties.get("Update"));
+ assertEquals(originalProperties.get("Update"), copyProperties.get("Update"));
+
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/ManagerSubscriberTest.java b/src/test/java/org/apache/sling/jobs/impl/ManagerSubscriberTest.java
new file mode 100644
index 0000000..7acc35c
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/ManagerSubscriberTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.jobs.impl;
+
+import aQute.libg.command.Command;
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobUpdate;
+import org.apache.sling.jobs.Types;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+import org.apache.sling.jobs.impl.storage.InMemoryJobStorage;
+import org.apache.sling.mom.QueueManager;
+import org.apache.sling.mom.TopicManager;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+
+import static org.junit.Assert.*;
+
+/**
+ */
+public class ManagerSubscriberTest {
+
+ private JobManagerImpl jobManager;
+ private JobStorage jobStorage;
+ private OutboundJobUpdateListener messageSender;
+ @Mock
+ private TopicManager topicManager;
+ @Mock
+ private QueueManager queueManager;
+ private HashMap<org.apache.sling.mom.Types.TopicName, Queue<QueueEntry>> topicQueues;
+ private HashMap<org.apache.sling.mom.Types.QueueName, Queue<QueueEntry>> messageQueues;
+ private ManagerSubscriber managerSubscriber;
+
+ public ManagerSubscriberTest() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ topicQueues = new HashMap<org.apache.sling.mom.Types.TopicName, Queue<QueueEntry>>();
+ messageQueues = new HashMap<org.apache.sling.mom.Types.QueueName, Queue<QueueEntry>>();
+ //noinspection unchecked
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ org.apache.sling.mom.Types.TopicName topic = (org.apache.sling.mom.Types.TopicName) invocationOnMock.getArguments()[0];
+ org.apache.sling.mom.Types.CommandName command = (org.apache.sling.mom.Types.CommandName) invocationOnMock.getArguments()[1];
+ @SuppressWarnings("unchecked") Map<String, Object> properties = (Map<String, Object>) invocationOnMock.getArguments()[2];
+ Queue<QueueEntry> queue = topicQueues.get(topic);
+ if (queue == null) {
+ queue = new ArrayBlockingQueue<QueueEntry>(100);
+ topicQueues.put(topic, queue);
+
+ }
+ queue.add(new QueueEntry(command, properties));
+ return null;
+ }
+ }).when(topicManager)
+ .publish(Mockito.any(org.apache.sling.mom.Types.TopicName.class), Mockito.any(org.apache.sling.mom.Types.CommandName.class), Mockito.any(Map.class));
+
+ //noinspection unchecked
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+ org.apache.sling.mom.Types.QueueName topic = (org.apache.sling.mom.Types.QueueName) invocationOnMock.getArguments()[0];
+ @SuppressWarnings("unchecked") Map<String, Object> properties = (Map<String, Object>) invocationOnMock.getArguments()[1];
+ Queue<QueueEntry> queue = messageQueues.get(topic);
+ if ( queue == null) {
+ queue = new ArrayBlockingQueue<QueueEntry>(100);
+ messageQueues.put(topic, queue);
+
+ }
+ queue.add(new QueueEntry(properties));
+ return null;
+ }
+ }).when(queueManager)
+ .add(Mockito.any(org.apache.sling.mom.Types.QueueName.class), Mockito.any(Map.class));
+
+ messageSender = new OutboundJobUpdateListener(topicManager, queueManager);
+ jobStorage = new InMemoryJobStorage();
+ jobManager = new JobManagerImpl(jobStorage, messageSender);
+
+ managerSubscriber = new ManagerSubscriber();
+ Field f = managerSubscriber.getClass().getDeclaredField("jobManager");
+ f.setAccessible(true);
+ f.set(managerSubscriber,jobManager);
+
+
+ }
+
+ @After
+ public void after() {
+ jobManager.dispose();
+ jobStorage.dispose();
+ }
+
+ @Test
+ public void test() {
+ // fake up a remote send.
+ Map<String, Object> properties = new HashMap<String, Object>();
+ String testId = "testGetJobById"+System.currentTimeMillis();
+ properties.put("testid", testId);
+ properties.put("job.name", "Jobname");
+ String jobId = Utils.generateId();
+ Job job = new JobImpl(Types.jobQueue("testtopic"), jobId,Types.jobType("testtype"), properties);
+ JobUpdateImpl jobUpdate = new JobUpdateImpl(job, JobUpdate.JobUpdateCommand.START_JOB, job.getProperties());
+
+ Map<String, Object> message = Utils.toMapValue(jobUpdate);
+
+ // pump the topic message into the managerSubscriber and check that the job can be found.
+
+ managerSubscriber.onMessage(org.apache.sling.mom.Types.topicName("testtopic"), message);
+
+ Job searchedJob = jobManager.getJobById(jobId);
+ assertNotNull(searchedJob);
+ assertEquals(job.getId(), searchedJob.getId());
+ assertEquals(testId, searchedJob.getProperties().get("testid"));
+
+ }
+
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/QueueEntry.java b/src/test/java/org/apache/sling/jobs/impl/QueueEntry.java
new file mode 100644
index 0000000..383dba4
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/QueueEntry.java
@@ -0,0 +1,48 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.mom.Types;
+
+import java.util.Map;
+
+/**
+ */
+class QueueEntry {
+ private final Types.CommandName name;
+ private final Map<String, Object> properties;
+
+ public QueueEntry(Types.CommandName name, Map<String, Object> properties) {
+ this.name = name;
+ this.properties = properties;
+ }
+
+ public QueueEntry(Map<String, Object> properties) {
+ this.name = null;
+ this.properties = properties;
+ }
+
+ public Types.CommandName getCommand() {
+ return name;
+ }
+
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+}
diff --git a/src/test/java/org/apache/sling/jobs/impl/UtilsTest.java b/src/test/java/org/apache/sling/jobs/impl/UtilsTest.java
new file mode 100644
index 0000000..a8d6e4c
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/UtilsTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.jobs.impl;
+
+import org.apache.sling.jobs.impl.spi.MapValueAdapter;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.Assert.*;
+
+/**
+ */
+public class UtilsTest {
+
+ @Test
+ public void testGenerateId() throws Exception {
+ // check the IDs don't clash. Cant check between separate JVMs without separate JVMs.
+ Set<String> ids = new HashSet<String>();
+ for ( int i = 0; i < 100; i++ ) {
+ ids.add(Utils.generateId());
+ }
+ assertEquals(100, ids.size());
+
+ }
+
+ @Test
+ public void testToMapValue() throws Exception {
+ final Map<String, Object> testMap = new HashMap<String, Object>();
+ testMap.put("v", "a");
+ Object mv = Utils.toMapValue(testMap);
+ assertEquals(testMap.size(), ((Map)mv).size());
+ assertEquals("a", ((Map)mv).get("v"));
+
+ mv = Utils.toMapValue(new MapValueAdapter() {
+ @Override
+ public void fromMapValue(Object mapValue) {
+
+ }
+
+ @Override
+ public Object toMapValue() {
+ return testMap;
+ }
+ });
+ assertEquals(testMap.size(), ((Map)mv).size());
+ assertEquals("a", ((Map)mv).get("v"));
+
+ try {
+ mv = Utils.toMapValue(new HashSet<String>());
+ fail("Should have rejected a hash set");
+ } catch ( IllegalArgumentException e) {
+ // ok
+ }
+
+ try {
+ mv = Utils.toMapValue(new MapValueAdapter() {
+ @Override
+ public void fromMapValue(Object mapValue) {
+
+ }
+
+ @Override
+ public Object toMapValue() {
+ return "should not be allowed as a MapValue is supposed to be a Map";
+ }
+ });
+ fail("Should have rejected a string from map Utils.toMapValue even if internal");
+ } catch ( IllegalArgumentException e) {
+ // ok
+ }
+
+ }
+
+ @Test
+ public void testGetRequired() throws Exception {
+ final Map<String, Object> testMap = new HashMap<String, Object>();
+ testMap.put("v", "a");
+ assertEquals("a",Utils.getRequired(testMap, "v"));
+ try {
+ Utils.getRequired(testMap, "z");
+ fail("Expected z to be missing and fail required test");
+ } catch ( IllegalArgumentException e) {
+ // ok
+ }
+ }
+
+ @Test
+ public void testGetOptional() throws Exception {
+ final Map<String, Object> testMap = new HashMap<String, Object>();
+ testMap.put("v", "a");
+ assertEquals("a", Utils.getOptional(testMap, "v", "xx"));
+ assertEquals("xx",Utils.getOptional(testMap, "z", "xx"));
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorageTest.java b/src/test/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorageTest.java
new file mode 100644
index 0000000..744c0e4
--- /dev/null
+++ b/src/test/java/org/apache/sling/jobs/impl/storage/InMemoryJobStorageTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.jobs.impl.storage;
+
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.impl.Utils;
+import org.apache.sling.jobs.impl.spi.JobStorage;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests in memory storage.
+ */
+public class InMemoryJobStorageTest {
+
+ private JobStorage jobStorage;
+
+ @Mock
+ private Job job;
+ private String jobId;
+
+ public InMemoryJobStorageTest() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ jobId = Utils.generateId();
+ Mockito.when(job.getId()).thenReturn(jobId);
+ jobStorage = new InMemoryJobStorage();
+ jobStorage.put(job);
+ }
+
+
+ @Test
+ public void testGet() throws Exception {
+ String nextJobId = Utils.generateId();
+ assertEquals(jobId, jobStorage.get(jobId).getId());
+ assertNull(jobStorage.get(nextJobId));
+ }
+
+ @Test
+ public void testPut() throws Exception {
+ String nextJobId = Utils.generateId();
+ Job newJob = Mockito.mock(Job.class);
+ Mockito.when(newJob.getId()).thenReturn(nextJobId);
+ jobStorage.put(newJob);
+
+ assertEquals(jobId, jobStorage.get(jobId).getId());
+ assertEquals(nextJobId, jobStorage.get(nextJobId).getId());
+ }
+
+ @Test
+ public void testRemove() throws Exception {
+ Job removed = jobStorage.remove(job);
+ assertNotNull(removed);
+ assertEquals(job.getId(), removed.getId());
+ assertNull(jobStorage.get(jobId));
+ }
+
+ @Test
+ public void testRemove1() throws Exception {
+ Job removed = jobStorage.remove(job.getId());
+ assertNotNull(removed);
+ assertEquals(job.getId(), removed.getId());
+ assertNull(jobStorage.get(jobId));
+
+ }
+}
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.
[sling-org-apache-sling-jobs] 05/06: [maven-release-plugin] prepare
release org.apache.sling.jobs-1.0.0
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit f74a7ad0d72e08459282c380e7c2cc6d20c65372
Author: Ian Boston <ie...@apache.org>
AuthorDate: Wed Oct 5 10:05:40 2016 +0000
[maven-release-plugin] prepare release org.apache.sling.jobs-1.0.0
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core@1763386 13f79535-47bb-0310-9956-ffa450edef68
---
pom.xml | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/pom.xml b/pom.xml
index 43b2911..97a9130 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,7 +28,7 @@
<artifactId>org.apache.sling.jobs</artifactId>
<packaging>bundle</packaging>
- <version>0.0.1-SNAPSHOT</version>
+ <version>1.0.0</version>
<name>Apache Sling Jobs Support</name>
<description>
@@ -36,9 +36,9 @@
</description>
<scm>
- <connection>scm:svn:http://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core</connection>
- <developerConnection>scm:svn:https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core</developerConnection>
- <url>http://svn.apache.org/viewvc/sling/trunk/contrib/commons/mom/jobs/core</url>
+ <connection>scm:svn:http://svn.apache.org/repos/asf/sling/tags/org.apache.sling.jobs-1.0.0</connection>
+ <developerConnection>scm:svn:https://svn.apache.org/repos/asf/sling/tags/org.apache.sling.jobs-1.0.0</developerConnection>
+ <url>http://svn.apache.org/viewvc/sling/tags/org.apache.sling.jobs-1.0.0</url>
</scm>
<properties>
@@ -82,7 +82,7 @@
<dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.mom</artifactId>
- <version>0.0.1-SNAPSHOT</version>
+ <version>1.0.0</version>
<scope>provided</scope>
</dependency>
<dependency>
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.
[sling-org-apache-sling-jobs] 03/06: SLING-5645 Improved IT tests,
improved config checking and reduced dependency footprint
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit 09609b06c5995ac52a88fc30e626593ce138b516
Author: Ian Boston <ie...@apache.org>
AuthorDate: Tue Oct 4 09:23:13 2016 +0000
SLING-5645 Improved IT tests, improved config checking and reduced dependency footprint
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core@1763255 13f79535-47bb-0310-9956-ffa450edef68
---
pom.xml | 5 ---
.../org/apache/sling/jobs/impl/JobSubsystem.java | 39 +++++++++++++++++++---
2 files changed, 35 insertions(+), 9 deletions(-)
diff --git a/pom.xml b/pom.xml
index 08a38ad..43b2911 100644
--- a/pom.xml
+++ b/pom.xml
@@ -116,11 +116,6 @@
<!-- Use the version being used in Sling -->
<version>15.0</version>
</dependency>
- <dependency>
- <groupId>org.apache.sling</groupId>
- <artifactId>org.apache.sling.commons.osgi</artifactId>
- <version>2.4.0</version>
- </dependency>
<!-- Testing -->
<dependency>
<groupId>junit</groupId>
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java b/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java
index ef3123b..1be6a00 100644
--- a/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java
+++ b/src/main/java/org/apache/sling/jobs/impl/JobSubsystem.java
@@ -28,8 +28,14 @@ import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.ReferencePolicy;
import org.apache.felix.scr.annotations.Service;
-import org.apache.sling.commons.osgi.PropertiesUtil;
-import org.apache.sling.jobs.*;
+import org.apache.sling.jobs.Job;
+import org.apache.sling.jobs.JobBuilder;
+import org.apache.sling.jobs.JobCallback;
+import org.apache.sling.jobs.JobConsumer;
+import org.apache.sling.jobs.JobManager;
+import org.apache.sling.jobs.JobTypeValve;
+import org.apache.sling.jobs.JobUpdateListener;
+import org.apache.sling.jobs.Types;
import org.apache.sling.jobs.impl.spi.JobStorage;
import org.apache.sling.jobs.impl.storage.InMemoryJobStorage;
import org.apache.sling.mom.QueueManager;
@@ -41,7 +47,8 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.Closeable;
-import java.util.Collection;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -180,10 +187,34 @@ public class JobSubsystem implements JobManager, JobConsumer {
if ( consumer instanceof JobTypeValve) {
jobTypes = ImmutableSet.of();
} else {
- jobTypes = Types.jobType(PropertiesUtil.toStringArray(properties.get(JobConsumer.JOB_TYPES)));
+ jobTypes = getJobTypes(properties);
}
}
+ public Set<Types.JobType> getJobTypes(Map<Object, Object> properties) {
+ Object types = properties.get(JobConsumer.JOB_TYPES);
+ if (types instanceof String) {
+ return Types.jobType(new String[]{(String) types});
+
+ } else if (types instanceof String[]) {
+ return Types.jobType((String[]) types);
+
+ } else if (types instanceof Iterable) {
+ List<String> l = new ArrayList<String>();
+ for (Object o : (Iterable<?>) types) {
+ l.add(String.valueOf(o));
+ }
+ return Types.jobType((String[]) l.toArray(new String[l.size()]));
+ }
+ throw new IllegalArgumentException("For the JobConsumer to work, the job consumer must either " +
+ "implement a JobTypeValve or define a list of JobTypes, neither were specified. " +
+ "Please check the implementation or OSGi configuration, was expecting " +
+ JobConsumer.JOB_TYPES + " property to be set to a String[]");
+ }
+
+
+
+
@Override
public boolean accept(@Nonnull Types.JobType jobType) {
if ( consumer instanceof JobTypeValve) {
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.
[sling-org-apache-sling-jobs] 02/06: SLING-5645 added provider
types to apis where applicable
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit 8e498cc9e19b6f49d6a2f28f77248b613fa1d406
Author: Ian Boston <ie...@apache.org>
AuthorDate: Mon Oct 3 16:17:56 2016 +0000
SLING-5645 added provider types to apis where applicable
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core@1763185 13f79535-47bb-0310-9956-ffa450edef68
---
pom.xml | 6 ++++++
src/main/java/org/apache/sling/jobs/Job.java | 2 ++
src/main/java/org/apache/sling/jobs/JobBuilder.java | 4 ++++
src/main/java/org/apache/sling/jobs/JobCallback.java | 4 ++++
src/main/java/org/apache/sling/jobs/JobConsumer.java | 4 ++++
src/main/java/org/apache/sling/jobs/JobController.java | 4 ++++
src/main/java/org/apache/sling/jobs/JobManager.java | 3 +++
src/main/java/org/apache/sling/jobs/JobTypeValve.java | 3 +++
src/main/java/org/apache/sling/jobs/JobUpdate.java | 2 ++
src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java | 3 +++
src/main/java/org/apache/sling/jobs/JobUpdateListener.java | 3 +++
src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java | 2 ++
src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java | 2 ++
src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java | 3 +++
14 files changed, 45 insertions(+)
diff --git a/pom.xml b/pom.xml
index 2343918..08a38ad 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,6 +100,12 @@
<scope>provided</scope>
</dependency>
<dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>osgi.annotation</artifactId>
+ <version>6.0.1</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
<version>2.0.1</version>
diff --git a/src/main/java/org/apache/sling/jobs/Job.java b/src/main/java/org/apache/sling/jobs/Job.java
index 819a7a2..e2dd33d 100644
--- a/src/main/java/org/apache/sling/jobs/Job.java
+++ b/src/main/java/org/apache/sling/jobs/Job.java
@@ -21,6 +21,7 @@ package org.apache.sling.jobs;
import com.google.common.collect.ImmutableSet;
import org.apache.sling.mom.TopicManager;
+import org.osgi.annotation.versioning.ProviderType;
import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
@@ -34,6 +35,7 @@ import java.util.Set;
* Largely based on org.apache.sling.event.Job.
* A Job contains the complete state of the Job as known in the current JVM.
*/
+@ProviderType
public interface Job {
diff --git a/src/main/java/org/apache/sling/jobs/JobBuilder.java b/src/main/java/org/apache/sling/jobs/JobBuilder.java
index b69934d..cabfd4f 100644
--- a/src/main/java/org/apache/sling/jobs/JobBuilder.java
+++ b/src/main/java/org/apache/sling/jobs/JobBuilder.java
@@ -18,12 +18,16 @@
*/
package org.apache.sling.jobs;
+
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
import java.util.Map;
/**
* A JobBuilder allows users of the JobSystem to modify the properties of the Job and submit it for processing.
*/
+@ProviderType
public interface JobBuilder {
/**
diff --git a/src/main/java/org/apache/sling/jobs/JobCallback.java b/src/main/java/org/apache/sling/jobs/JobCallback.java
index bbadc6b..4fc84e3 100644
--- a/src/main/java/org/apache/sling/jobs/JobCallback.java
+++ b/src/main/java/org/apache/sling/jobs/JobCallback.java
@@ -19,10 +19,14 @@
package org.apache.sling.jobs;
+
+import org.osgi.annotation.versioning.ProviderType;
+
/**
* When a job is complete, the callback gets called.
* This interface is implemented by the Jobs Implementation and passed into a JobsConsumer.
*/
+@ProviderType
public interface JobCallback {
/**
diff --git a/src/main/java/org/apache/sling/jobs/JobConsumer.java b/src/main/java/org/apache/sling/jobs/JobConsumer.java
index dbf6703..af808db 100644
--- a/src/main/java/org/apache/sling/jobs/JobConsumer.java
+++ b/src/main/java/org/apache/sling/jobs/JobConsumer.java
@@ -18,6 +18,9 @@
*/
package org.apache.sling.jobs;
+
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
/**
@@ -30,6 +33,7 @@ import javax.annotation.Nonnull;
* the job queue is to be discouraged, as it may require too much from the JobConsumer implementation to guarantee the contract
* in this interface.
*/
+@ProviderType
public interface JobConsumer {
String JOB_TYPES = "job.types";
diff --git a/src/main/java/org/apache/sling/jobs/JobController.java b/src/main/java/org/apache/sling/jobs/JobController.java
index e038b52..153ad62 100644
--- a/src/main/java/org/apache/sling/jobs/JobController.java
+++ b/src/main/java/org/apache/sling/jobs/JobController.java
@@ -19,6 +19,9 @@
package org.apache.sling.jobs;
+
+import org.osgi.annotation.versioning.ProviderType;
+
/**
* A job controller provides a mechanism by which a running job can be sent control messages.
* Jobs when they start should register a JobController implementation with the JobManager so that
@@ -26,6 +29,7 @@ package org.apache.sling.jobs;
* message to the Job, bypassing the manager, although this TopicListener will need to filter
* for all messages associated with the Job.
*/
+@ProviderType
public interface JobController {
diff --git a/src/main/java/org/apache/sling/jobs/JobManager.java b/src/main/java/org/apache/sling/jobs/JobManager.java
index 5dc4c3c..e23db30 100644
--- a/src/main/java/org/apache/sling/jobs/JobManager.java
+++ b/src/main/java/org/apache/sling/jobs/JobManager.java
@@ -18,6 +18,8 @@
*/
package org.apache.sling.jobs;
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Collection;
@@ -26,6 +28,7 @@ import java.util.Map;
/**
* Based on org.apache.sling.event.job.JobManager but trimmed down and adjusted to deal with distributed job queues.
*/
+@ProviderType
public interface JobManager {
diff --git a/src/main/java/org/apache/sling/jobs/JobTypeValve.java b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
index 20f6782..2b75599 100644
--- a/src/main/java/org/apache/sling/jobs/JobTypeValve.java
+++ b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
@@ -18,6 +18,8 @@
*/
package org.apache.sling.jobs;
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
/**
@@ -25,6 +27,7 @@ import javax.annotation.Nonnull;
* it can perform further actions on it. JobTypeValves are used in place of static JobType declarations.
* This interface is only implemented where static declaration of JobTypes does not satisfy the use cases.
*/
+@ProviderType
public interface JobTypeValve {
/**
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdate.java b/src/main/java/org/apache/sling/jobs/JobUpdate.java
index 62723ad..333fcd9 100644
--- a/src/main/java/org/apache/sling/jobs/JobUpdate.java
+++ b/src/main/java/org/apache/sling/jobs/JobUpdate.java
@@ -19,6 +19,7 @@
package org.apache.sling.jobs;
import org.apache.sling.mom.TopicManager;
+import org.osgi.annotation.versioning.ProviderType;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -36,6 +37,7 @@ import java.util.Map;
* The API does not specify the implementation, only that the various JobUpdate messages are delivered with an delay appropriate for the
* type of message, identified by the JobUpdateCommand type.
*/
+@ProviderType
public interface JobUpdate {
/**
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java b/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java
index d691348..e4c938d 100644
--- a/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java
+++ b/src/main/java/org/apache/sling/jobs/JobUpdateBuilder.java
@@ -19,6 +19,8 @@
package org.apache.sling.jobs;
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Map;
@@ -26,6 +28,7 @@ import java.util.Map;
/**
* Create a job update ading properties and building.
*/
+@ProviderType
public interface JobUpdateBuilder {
@Nonnull
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdateListener.java b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
index 82a2c63..7cbaa59 100644
--- a/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
+++ b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
@@ -18,12 +18,15 @@
*/
package org.apache.sling.jobs;
+import org.osgi.annotation.versioning.ProviderType;
+
import javax.annotation.Nonnull;
/**
* Listens for JobUpdate messages.
* Examples of a JobUpdateListener include implementations of a Job that consumes JobUpdateMessages, or JobConsumers that execute a Job.
*/
+@ProviderType
public interface JobUpdateListener {
/**
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java b/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java
index f9ccd76..eedbbd6 100644
--- a/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/JobStarter.java
@@ -19,10 +19,12 @@
package org.apache.sling.jobs.impl.spi;
import org.apache.sling.jobs.Job;
+import org.osgi.annotation.versioning.ProviderType;
/**
* Starts a job, used by the JobBuilderImpl to perform the add operation.
*/
+@ProviderType
public interface JobStarter {
Job start(Job j);
}
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
index e33968f..881e30b 100644
--- a/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
@@ -20,6 +20,7 @@
package org.apache.sling.jobs.impl.spi;
import org.apache.sling.jobs.Job;
+import org.osgi.annotation.versioning.ProviderType;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -27,6 +28,7 @@ import javax.annotation.Nullable;
/**
* Provides JobStorage local to the JVM. Implementation may or may not decide to persist over restarts, page, etc
*/
+@ProviderType
public interface JobStorage {
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
index 2266ee6..54a9ba2 100644
--- a/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
@@ -18,12 +18,15 @@
*/
package org.apache.sling.jobs.impl.spi;
+import org.osgi.annotation.versioning.ProviderType;
+
/**
* Objects that can be converted to and from a Map are expected to extend this base class.
* The values contained in the Map are expected to be Maps or values that can be serialised into most
* common formats. It would be safe to use json or yaml as an example of a common format.
*
*/
+@ProviderType
public interface MapValueAdapter {
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.
[sling-org-apache-sling-jobs] 04/06: SLING-5645 fixes to javadoc
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to annotated tag org.apache.sling.jobs-1.0.0
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-jobs.git
commit d4a02256f4837f90c49bffbb7055a85c9dc34295
Author: Ian Boston <ie...@apache.org>
AuthorDate: Wed Oct 5 09:21:31 2016 +0000
SLING-5645 fixes to javadoc
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/jobs/core@1763369 13f79535-47bb-0310-9956-ffa450edef68
---
src/main/java/org/apache/sling/jobs/Job.java | 11 ++++++++++-
src/main/java/org/apache/sling/jobs/JobCallback.java | 2 +-
src/main/java/org/apache/sling/jobs/JobManager.java | 1 +
src/main/java/org/apache/sling/jobs/JobTypeValve.java | 2 +-
src/main/java/org/apache/sling/jobs/JobUpdate.java | 7 +++++--
src/main/java/org/apache/sling/jobs/JobUpdateListener.java | 2 +-
src/main/java/org/apache/sling/jobs/Types.java | 8 ++++----
src/main/java/org/apache/sling/jobs/impl/JobImpl.java | 2 +-
src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java | 4 ++--
.../java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java | 2 +-
src/main/java/org/apache/sling/jobs/impl/Utils.java | 2 +-
src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java | 2 +-
.../java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java | 4 ++--
13 files changed, 31 insertions(+), 18 deletions(-)
diff --git a/src/main/java/org/apache/sling/jobs/Job.java b/src/main/java/org/apache/sling/jobs/Job.java
index e2dd33d..284f346 100644
--- a/src/main/java/org/apache/sling/jobs/Job.java
+++ b/src/main/java/org/apache/sling/jobs/Job.java
@@ -86,18 +86,21 @@ public interface Job {
/**
* On first execution the value of this property is zero.
* This property is managed by the job handling.
+ * @return the retry count.
*/
int getRetryCount();
/**
* The property to track the retry maximum retry count for jobs.
* This property is managed by the job handling.
+ * @return the number of retries.
*/
int getNumberOfRetries();
/**
* The time when the job started.
+ * @return the epch time in ms when the job was started.
*/
long getStarted();
@@ -109,13 +112,14 @@ public interface Job {
/**
* Get the job state
+ * @return the job state.
*/
@Nonnull
JobState getJobState();
/**
* Set the new state.
- * @param newState
+ * @param newState the new state.
*/
void setState(@Nonnull JobState newState);
@@ -135,6 +139,10 @@ public interface Job {
@Nullable
String getResultMessage();
+ /**
+ *
+ * @return a Job update builder.
+ */
@Nonnull
JobUpdateBuilder newJobUpdateBuilder();
@@ -148,6 +156,7 @@ public interface Job {
/**
* Set the current job controller.
+ * @param jobController the job controller.
*/
void setJobController(@Nonnull JobController jobController);
diff --git a/src/main/java/org/apache/sling/jobs/JobCallback.java b/src/main/java/org/apache/sling/jobs/JobCallback.java
index 4fc84e3..4348e47 100644
--- a/src/main/java/org/apache/sling/jobs/JobCallback.java
+++ b/src/main/java/org/apache/sling/jobs/JobCallback.java
@@ -31,7 +31,7 @@ public interface JobCallback {
/**
* Call back with an update job state.
- * @param finalJobState
+ * @param finalJobState job state.
*/
void callback(Job finalJobState);
}
diff --git a/src/main/java/org/apache/sling/jobs/JobManager.java b/src/main/java/org/apache/sling/jobs/JobManager.java
index e23db30..1354342 100644
--- a/src/main/java/org/apache/sling/jobs/JobManager.java
+++ b/src/main/java/org/apache/sling/jobs/JobManager.java
@@ -139,6 +139,7 @@ public interface JobManager {
* When a job is stopped and the job consumer supports stopping the job processing, it is up
* to the job consumer how the stopping is handled. The job can be marked as finished successful,
* permanently failed or being retried.
+ * @param jobId the job id to stop.
*/
void stopJobById(@Nonnull String jobId);
diff --git a/src/main/java/org/apache/sling/jobs/JobTypeValve.java b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
index 2b75599..3be4f34 100644
--- a/src/main/java/org/apache/sling/jobs/JobTypeValve.java
+++ b/src/main/java/org/apache/sling/jobs/JobTypeValve.java
@@ -32,7 +32,7 @@ public interface JobTypeValve {
/**
* Return true if the component, normally a JobConsumer, can process the jobType.
- * @param jobType
+ * @param jobType the job type.
* @return true if can be processed.
*/
boolean accept(@Nonnull Types.JobType jobType);
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdate.java b/src/main/java/org/apache/sling/jobs/JobUpdate.java
index 333fcd9..301cf47 100644
--- a/src/main/java/org/apache/sling/jobs/JobUpdate.java
+++ b/src/main/java/org/apache/sling/jobs/JobUpdate.java
@@ -33,7 +33,7 @@ import java.util.Map;
* put the JobUpdates that start a message on a separate queue from the JobUpdates that update, abort or stop a job, so that
* those messages do not get delayed by a backlog of job start messages. Alternatively the implementation may decide to implement
* a scheduler that consumes job start messages at full queue throughput and queue jobs into a separate implementation specific queue.
- * <p/>
+ * <p>
* The API does not specify the implementation, only that the various JobUpdate messages are delivered with an delay appropriate for the
* type of message, identified by the JobUpdateCommand type.
*/
@@ -56,7 +56,7 @@ public interface JobUpdate {
* exist on the receiver the update message may be stored until the TTL expires before it is applied. This allows
* messages sent out of order to wait till expires has been reached before applying the update. The rules determining
* how expiring messages are applied is an implementation detail.
- * @return
+ * @return the epoch time in ms when the update expires.
*/
long expires();
@@ -125,18 +125,21 @@ public interface JobUpdate {
/**
* On first execution the value of this property is zero.
* This property is managed by the job handling.
+ * @return the retry count.
*/
int getRetryCount();
/**
* The property to track the retry maximum retry count for jobs.
* This property is managed by the job handling.
+ * @return the number of retries.
*/
int getNumberOfRetries();
/**
* The time when the job started.
+ * @return epoch time when the job was started.
*/
long getStarted();
diff --git a/src/main/java/org/apache/sling/jobs/JobUpdateListener.java b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
index 7cbaa59..3e6e90d 100644
--- a/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
+++ b/src/main/java/org/apache/sling/jobs/JobUpdateListener.java
@@ -31,7 +31,7 @@ public interface JobUpdateListener {
/**
* Update the job state with a message.
- * @param update
+ * @param update update message.
*/
void update(@Nonnull JobUpdate update);
}
diff --git a/src/main/java/org/apache/sling/jobs/Types.java b/src/main/java/org/apache/sling/jobs/Types.java
index 00f6501..da7c763 100644
--- a/src/main/java/org/apache/sling/jobs/Types.java
+++ b/src/main/java/org/apache/sling/jobs/Types.java
@@ -56,8 +56,8 @@ public final class Types {
/**
* Create a JobQueue from a string represtnation, normally this us a path representing the jobQueue, although the path
* hierachy may not have any significance depending on the MoM implementation.
- * @param jobQueue
- * @return
+ * @param jobQueue job queue name.
+ * @return the job queue.
*/
public static JobQueue jobQueue(String jobQueue) {
return new JobQueueImpl(jobQueue);
@@ -65,8 +65,8 @@ public final class Types {
/**
* Create a JobType from a string representation.
- * @param jobType
- * @return
+ * @param jobType job type name.
+ * @return job type.
*/
public static JobType jobType(String jobType) {
return new JobTypeImpl(jobType);
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobImpl.java
index ebc0971..6ece8f1 100644
--- a/src/main/java/org/apache/sling/jobs/impl/JobImpl.java
+++ b/src/main/java/org/apache/sling/jobs/impl/JobImpl.java
@@ -147,7 +147,7 @@ public class JobImpl implements Job, JobUpdateListener {
/**
* Apply an job update to this job, checking that the update is valid for the job.
- * @param jobUpdate
+ * @param jobUpdate job update.
*/
@Override
public void update(@Nonnull JobUpdate jobUpdate) {
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java
index 5d9ff4b..58bb464 100644
--- a/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java
+++ b/src/main/java/org/apache/sling/jobs/impl/JobManagerImpl.java
@@ -49,8 +49,8 @@ public class JobManagerImpl implements JobManager, JobStarter, JobUpdateListene
/**
*
- * @param jobStorage
- * @param messageSender
+ * @param jobStorage job storage used by the manager.
+ * @param messageSender the message sender that listens for updates.
*/
public JobManagerImpl(JobStorage jobStorage, JobUpdateListener messageSender) {
this.jobStorage = jobStorage;
diff --git a/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java b/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java
index 028dc3b..f708c25 100644
--- a/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java
+++ b/src/main/java/org/apache/sling/jobs/impl/JobUpdateBuilderImpl.java
@@ -38,7 +38,7 @@ public class JobUpdateBuilderImpl implements JobUpdateBuilder {
/**
* Create a JobUpdateBuilder from a job.
- * @param job
+ * @param job the job.
*/
public JobUpdateBuilderImpl(@Nonnull Job job) {
this.job = job;
diff --git a/src/main/java/org/apache/sling/jobs/impl/Utils.java b/src/main/java/org/apache/sling/jobs/impl/Utils.java
index 53d4475..662d002 100644
--- a/src/main/java/org/apache/sling/jobs/impl/Utils.java
+++ b/src/main/java/org/apache/sling/jobs/impl/Utils.java
@@ -97,7 +97,7 @@ public class Utils {
/**
* Generate an ID based on the unique name of the jvm process and a counter.
- * @return
+ * @return a generated ID.
*/
@Nonnull
public static String generateId() {
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
index 881e30b..bd421ab 100644
--- a/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/JobStorage.java
@@ -50,7 +50,7 @@ public interface JobStorage {
/**
* Remove the Job
- * @param jobId
+ * @param jobId the job id.
* @return the job removed or null if not present.
*/
@Nullable
diff --git a/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
index 54a9ba2..2018243 100644
--- a/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
+++ b/src/main/java/org/apache/sling/jobs/impl/spi/MapValueAdapter.java
@@ -32,12 +32,12 @@ public interface MapValueAdapter {
/**
* Populate the object from a map value.
- * @param mapValue
+ * @param mapValue object representing the map value.
*/
void fromMapValue(Object mapValue);
/**
- * Adapt the object into a value suitable for use in a map to be serialised by standard map -> json,yaml writers.
+ * Adapt the object into a value suitable for use in a map to be serialised by standard map to json,yaml writers.
* @return a value, which may be a primitive, an array or a map of primitives.
*/
Object toMapValue();
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.