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/10/20 14:37:51 UTC
[sling-org-apache-sling-event] 04/26: SLING-6824 : Migrate to R6
annotations, clean up dependencies
This is an automated email from the ASF dual-hosted git repository.
rombert pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-event.git
commit 35f9a0da03c5ff1d178eca587874b76406910239
Author: Carsten Ziegeler <cz...@apache.org>
AuthorDate: Thu May 4 08:03:38 2017 +0000
SLING-6824 : Migrate to R6 annotations, clean up dependencies
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk@1793745 13f79535-47bb-0310-9956-ffa450edef68
---
pom.xml | 17 +-
.../sling/event/impl/EnvironmentComponent.java | 19 +-
.../sling/event/impl/EventingThreadPool.java | 53 +++---
.../sling/event/impl/jobs/JobConsumerManager.java | 110 +++++------
.../sling/event/impl/jobs/JobManagerImpl.java | 35 ++--
.../org/apache/sling/event/impl/jobs/Utility.java | 19 +-
.../impl/jobs/config/ConfigurationConstants.java | 4 -
.../jobs/config/InternalQueueConfiguration.java | 204 +++++++++++----------
.../impl/jobs/config/JobManagerConfiguration.java | 110 +++++------
.../impl/jobs/config/MainQueueConfiguration.java | 178 ++++++++++++------
.../jobs/config/QueueConfigurationManager.java | 24 +--
.../event/impl/jobs/config/TopologyHandler.java | 18 +-
.../event/impl/jobs/console/InventoryPlugin.java | 25 ++-
.../event/impl/jobs/console/WebConsolePlugin.java | 22 +--
.../event/impl/jobs/jmx/AllJobStatisticsMBean.java | 17 +-
.../sling/event/impl/jobs/jmx/QueuesMBeanImpl.java | 26 +--
.../impl/jobs/notifications/NewJobSender.java | 14 +-
.../sling/event/impl/jobs/queues/QueueManager.java | 40 ++--
.../event/impl/jobs/stats/StatisticsManager.java | 16 +-
.../event/impl/jobs/tasks/HistoryCleanUpTask.java | 17 +-
.../event/impl/jobs/JobConsumerManagerTest.java | 40 +++-
.../config/InternalQueueConfigurationTest.java | 160 ++++++++++------
.../sling/event/it/AbstractJobHandlingTest.java | 10 +-
23 files changed, 636 insertions(+), 542 deletions(-)
diff --git a/pom.xml b/pom.xml
index 641a79b..3f81be8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -205,6 +205,10 @@
</profiles>
<dependencies>
<dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.service.component.annotations</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.discovery.api</artifactId>
<version>1.0.0</version>
@@ -223,8 +227,6 @@
<dependency>
<groupId>org.osgi</groupId>
<artifactId>osgi.core</artifactId>
- <version>6.0.0</version>
- <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.osgi</groupId>
@@ -233,11 +235,6 @@
<scope>provided</scope>
</dependency>
<dependency>
- <groupId>org.apache.felix</groupId>
- <artifactId>org.apache.felix.scr.annotations</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.settings</artifactId>
<version>1.0.0</version>
@@ -316,13 +313,13 @@
<dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.testing.tools</artifactId>
- <version>1.0.6</version>
+ <version>1.0.14</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.testing.sling-mock</artifactId>
- <version>1.6.0</version>
+ <version>2.2.6</version>
<scope>test</scope>
</dependency>
@@ -364,7 +361,7 @@
<dependency>
<groupId>org.apache.felix</groupId>
<artifactId>org.apache.felix.framework</artifactId>
- <version>5.4.0</version>
+ <version>5.6.2</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/src/main/java/org/apache/sling/event/impl/EnvironmentComponent.java b/src/main/java/org/apache/sling/event/impl/EnvironmentComponent.java
index b3f7929..7c3caad 100644
--- a/src/main/java/org/apache/sling/event/impl/EnvironmentComponent.java
+++ b/src/main/java/org/apache/sling/event/impl/EnvironmentComponent.java
@@ -18,14 +18,14 @@
*/
package org.apache.sling.event.impl;
-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.Service;
import org.apache.sling.commons.threads.ThreadPool;
import org.apache.sling.event.impl.support.Environment;
import org.apache.sling.settings.SlingSettingsService;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
/**
* Environment component. This component provides "global settings"
@@ -35,14 +35,17 @@ import org.apache.sling.settings.SlingSettingsService;
* This component needs to be immediate to set the global variables
* (application id and thread pool).
*/
-@Component(immediate=true)
-@Service(value=EnvironmentComponent.class)
+@Component(immediate=true,
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ },
+ service={EnvironmentComponent.class})
public class EnvironmentComponent {
/**
* Our thread pool.
*/
- @Reference(referenceInterface=EventingThreadPool.class)
+ @Reference(service=EventingThreadPool.class)
private ThreadPool threadPool;
/** Sling settings service. */
diff --git a/src/main/java/org/apache/sling/event/impl/EventingThreadPool.java b/src/main/java/org/apache/sling/event/impl/EventingThreadPool.java
index 0821866..652f09f 100644
--- a/src/main/java/org/apache/sling/event/impl/EventingThreadPool.java
+++ b/src/main/java/org/apache/sling/event/impl/EventingThreadPool.java
@@ -18,34 +18,44 @@
*/
package org.apache.sling.event.impl;
-import java.util.Map;
-
-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.Modified;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
-import org.apache.sling.commons.osgi.PropertiesUtil;
import org.apache.sling.commons.threads.ModifiableThreadPoolConfig;
import org.apache.sling.commons.threads.ThreadPool;
import org.apache.sling.commons.threads.ThreadPoolConfig;
import org.apache.sling.commons.threads.ThreadPoolConfig.ThreadPriority;
import org.apache.sling.commons.threads.ThreadPoolManager;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
/**
* The configurable eventing thread pool.
*/
-@Component(label="Apache Sling Job Thread Pool",
+@Component(service = EventingThreadPool.class,
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
+@Designate(ocd = EventingThreadPool.Config.class)
+public class EventingThreadPool implements ThreadPool {
+
+ @ObjectClassDefinition(name = "Apache Sling Job Thread Pool",
description="This is the thread pool used by the Apache Sling job handling. The "
+ "threads from this pool are merely used for executing jobs. By limiting this pool, it is "
+ "possible to limit the maximum number of parallel processed jobs - regardless of the queue "
- + "configuration.",
- metatype=true)
-@Service(value=EventingThreadPool.class)
-public class EventingThreadPool implements ThreadPool {
+ + "configuration.")
+ public @interface Config {
+
+ @AttributeDefinition(name = "Pool Size",
+ description="The size of the thread pool. This pool is used to execute jobs and therefore "
+ + "limits the maximum number of jobs executed in parallel.")
+ int minPoolSize() default 35;
+ }
@Reference
private ThreadPoolManager threadPoolManager;
@@ -53,14 +63,6 @@ public class EventingThreadPool implements ThreadPool {
/** The real thread pool used. */
private org.apache.sling.commons.threads.ThreadPool threadPool;
- private static final int DEFAULT_POOL_SIZE = 35;
-
- @Property(intValue=DEFAULT_POOL_SIZE,
- label="Pool Size",
- description="The size of the thread pool. This pool is used to execute jobs and therefore "
- + "limits the maximum number of jobs executed in parallel.")
- private static final String PROPERTY_POOL_SIZE = "minPoolSize";
-
public EventingThreadPool() {
// default constructor
}
@@ -79,9 +81,8 @@ public class EventingThreadPool implements ThreadPool {
*/
@Activate
@Modified
- protected void activate(final Map<String, Object> props) {
- final int maxPoolSize = PropertiesUtil.toInteger(props.get(PROPERTY_POOL_SIZE), DEFAULT_POOL_SIZE);
- this.configure(maxPoolSize);
+ protected void activate(final Config config) {
+ this.configure(config.minPoolSize());
}
private void configure(final int maxPoolSize) {
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/JobConsumerManager.java b/src/main/java/org/apache/sling/event/impl/jobs/JobConsumerManager.java
index 08660e8..5aa4eb3 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/JobConsumerManager.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/JobConsumerManager.java
@@ -27,17 +27,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-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.Modified;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.PropertyUnbounded;
-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.References;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.commons.osgi.PropertiesUtil;
import org.apache.sling.discovery.PropertyProvider;
import org.apache.sling.event.impl.jobs.config.TopologyCapabilities;
@@ -53,55 +42,62 @@ import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.osgi.service.component.annotations.ReferencePolicy;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This component manages/keeps track of all job consumer services.
*/
-@Component(label="Apache Sling Job Consumer Manager",
+@Component(service = JobConsumerManager.class,
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ })
+@Designate(ocd = JobConsumerManager.Config.class)
+public class JobConsumerManager {
+
+ @ObjectClassDefinition(name = "Apache Sling Job Consumer Manager",
description="The consumer manager controls the job consumer (= processors). "
+ "It can be used to temporarily disable job processing on the current instance. Other instances "
- + "in a cluster are not affected.",
- metatype=true)
-@Service(value=JobConsumerManager.class)
-@References({
- @Reference(referenceInterface=JobConsumer.class,
- cardinality=ReferenceCardinality.OPTIONAL_MULTIPLE,
- policy=ReferencePolicy.DYNAMIC),
- @Reference(referenceInterface=JobExecutor.class,
- cardinality=ReferenceCardinality.OPTIONAL_MULTIPLE,
- policy=ReferencePolicy.DYNAMIC)
-})
-@Property(name="org.apache.sling.installer.configuration.persist", boolValue=false,
- label="Distribute config",
+ + "in a cluster are not affected.")
+ public @interface Config {
+
+ @AttributeDefinition(name = "Distribute config",
description="If this is disabled, the configuration is not persisted on save in the cluster and is "
+ "only used on the current instance. This option should always be disabled!")
-public class JobConsumerManager {
-
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ boolean org_apache_sling_installer_configuration_persist() default false;
- @Property(unbounded=PropertyUnbounded.ARRAY, value = "*",
- label="Topic Whitelist",
+ @AttributeDefinition(name = "Topic Whitelist",
description="This is a list of topics which currently should be "
+ "processed by this instance. Leaving it empty, all job consumers are disabled. Putting a '*' as "
+ "one entry, enables all job consumers. Adding separate topics enables job consumers for exactly "
+ "this topic.")
- private static final String PROPERTY_WHITELIST = "job.consumermanager.whitelist";
+ String[] job_consumermanager_whitelist() default "*";
- @Property(unbounded=PropertyUnbounded.ARRAY,
- label="Topic Blacklist",
+ @AttributeDefinition(name = "Topic Blacklist",
description="This is a list of topics which currently shouldn't be "
+ "processed by this instance. Leaving it empty, all job consumers are enabled. Putting a '*' as "
+ "one entry, disables all job consumers. Adding separate topics disables job consumers for exactly "
+ "this topic.")
- private static final String PROPERTY_BLACKLIST = "job.consumermanager.blacklist";
+ String[] job_consumermanager_blacklist();
+ }
+
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
/** The map with the consumers, keyed by topic, sorted by service ranking. */
- private final Map<String, List<ConsumerInfo>> topicToConsumerMap = new HashMap<String, List<ConsumerInfo>>();
+ private final Map<String, List<ConsumerInfo>> topicToConsumerMap = new HashMap<>();
/** ServiceRegistration for propagation. */
- private ServiceRegistration propagationService;
+ private ServiceRegistration<PropertyProvider> propagationService;
private String topics;
@@ -113,10 +109,10 @@ public class JobConsumerManager {
private BundleContext bundleContext;
- private final Map<String, Object[]> listenerMap = new HashMap<String, Object[]>();
+ private final Map<String, Object[]> listenerMap = new HashMap<>();
private Dictionary<String, Object> getRegistrationProperties() {
- final Dictionary<String, Object> serviceProps = new Hashtable<String, Object>();
+ final Dictionary<String, Object> serviceProps = new Hashtable<>();
serviceProps.put(PropertyProvider.PROPERTY_PROPERTIES, TopologyCapabilities.PROPERTY_TOPICS);
// we add a changing property to the service registration
// to make sure a modification event is really sent
@@ -127,16 +123,16 @@ public class JobConsumerManager {
}
@Activate
- protected void activate(final BundleContext bc, final Map<String, Object> props) {
+ protected void activate(final BundleContext bc, final Config config) {
this.bundleContext = bc;
- this.modified(bc, props);
+ this.modified(bc, config);
}
@Modified
- protected void modified(final BundleContext bc, final Map<String, Object> props) {
+ protected void modified(final BundleContext bc, final Config config) {
final boolean wasEnabled = this.propagationService != null;
- this.whitelistMatchers = TopicMatcherHelper.buildMatchers(PropertiesUtil.toStringArray(props.get(PROPERTY_WHITELIST)));
- this.blacklistMatchers = TopicMatcherHelper.buildMatchers(PropertiesUtil.toStringArray(props.get(PROPERTY_BLACKLIST)));
+ this.whitelistMatchers = TopicMatcherHelper.buildMatchers(config.job_consumermanager_whitelist());
+ this.blacklistMatchers = TopicMatcherHelper.buildMatchers(config.job_consumermanager_blacklist());
final boolean enable = this.whitelistMatchers != null && this.blacklistMatchers != TopicMatcherHelper.MATCH_ALL;
if ( wasEnabled != enable ) {
@@ -145,7 +141,7 @@ public class JobConsumerManager {
}
if ( enable ) {
logger.debug("Registering property provider with: {}", this.topics);
- this.propagationService = bc.registerService(PropertyProvider.class.getName(),
+ this.propagationService = bc.registerService(PropertyProvider.class,
new PropertyProvider() {
@Override
@@ -240,7 +236,10 @@ public class JobConsumerManager {
* Bind a new consumer
* @param serviceReference The service reference to the consumer.
*/
- protected void bindJobConsumer(final ServiceReference serviceReference) {
+ @Reference(service=JobConsumer.class,
+ cardinality=ReferenceCardinality.MULTIPLE,
+ policy=ReferencePolicy.DYNAMIC)
+ protected void bindJobConsumer(final ServiceReference<JobConsumer> serviceReference) {
this.bindService(serviceReference, true);
}
@@ -248,7 +247,7 @@ public class JobConsumerManager {
* Unbind a consumer
* @param serviceReference The service reference to the consumer.
*/
- protected void unbindJobConsumer(final ServiceReference serviceReference) {
+ protected void unbindJobConsumer(final ServiceReference<JobConsumer> serviceReference) {
this.unbindService(serviceReference, true);
}
@@ -256,7 +255,10 @@ public class JobConsumerManager {
* Bind a new executor
* @param serviceReference The service reference to the executor.
*/
- protected void bindJobExecutor(final ServiceReference serviceReference) {
+ @Reference(service=JobExecutor.class,
+ cardinality=ReferenceCardinality.MULTIPLE,
+ policy=ReferencePolicy.DYNAMIC)
+ protected void bindJobExecutor(final ServiceReference<JobExecutor> serviceReference) {
this.bindService(serviceReference, false);
}
@@ -264,7 +266,7 @@ public class JobConsumerManager {
* Unbind a executor
* @param serviceReference The service reference to the executor.
*/
- protected void unbindJobExecutor(final ServiceReference serviceReference) {
+ protected void unbindJobExecutor(final ServiceReference<JobExecutor> serviceReference) {
this.unbindService(serviceReference, false);
}
@@ -273,7 +275,7 @@ public class JobConsumerManager {
* @param serviceReference The service reference to the consumer or executor.
* @param isConsumer Indicating whether this is a JobConsumer or JobExecutor
*/
- private void bindService(final ServiceReference serviceReference, final boolean isConsumer) {
+ private void bindService(final ServiceReference<?> serviceReference, final boolean isConsumer) {
final String[] topics = PropertiesUtil.toStringArray(serviceReference.getProperty(JobConsumer.PROPERTY_TOPICS));
if ( topics != null && topics.length > 0 ) {
final ConsumerInfo info = new ConsumerInfo(serviceReference, isConsumer);
@@ -285,7 +287,7 @@ public class JobConsumerManager {
if ( topic.length() > 0 ) {
List<ConsumerInfo> consumers = this.topicToConsumerMap.get(topic);
if ( consumers == null ) {
- consumers = new ArrayList<JobConsumerManager.ConsumerInfo>();
+ consumers = new ArrayList<>();
this.topicToConsumerMap.put(topic, consumers);
changed = true;
}
@@ -310,7 +312,7 @@ public class JobConsumerManager {
* @param serviceReference The service reference to the consumer or executor.
* @param isConsumer Indicating whether this is a JobConsumer or JobExecutor
*/
- private void unbindService(final ServiceReference serviceReference, final boolean isConsumer) {
+ private void unbindService(final ServiceReference<?> serviceReference, final boolean isConsumer) {
final String[] topics = PropertiesUtil.toStringArray(serviceReference.getProperty(JobConsumer.PROPERTY_TOPICS));
if ( topics != null && topics.length > 0 ) {
final ConsumerInfo info = new ConsumerInfo(serviceReference, isConsumer);
@@ -367,7 +369,7 @@ public class JobConsumerManager {
if ( enabled ) {
// create a sorted list - this ensures that the property value
// is always the same for the same topics.
- final List<String> topicList = new ArrayList<String>();
+ final List<String> topicList = new ArrayList<>();
for(final String topic : this.topicToConsumerMap.keySet() ) {
// check whitelist
if ( this.match(topic, this.whitelistMatchers) ) {
@@ -400,13 +402,13 @@ public class JobConsumerManager {
*/
private final static class ConsumerInfo implements Comparable<ConsumerInfo> {
- public final ServiceReference serviceReference;
+ public final ServiceReference<?> serviceReference;
private final boolean isConsumer;
public JobExecutor executor;
public final int ranking;
public final long serviceId;
- public ConsumerInfo(final ServiceReference serviceReference, final boolean isConsumer) {
+ public ConsumerInfo(final ServiceReference<?> serviceReference, final boolean isConsumer) {
this.serviceReference = serviceReference;
this.isConsumer = isConsumer;
final Object sr = serviceReference.getProperty(Constants.SERVICE_RANKING);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java b/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java
index b76fda4..b5c6372 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java
@@ -28,13 +28,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
-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.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.jackrabbit.util.ISO9075;
import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.PersistenceException;
@@ -69,6 +62,10 @@ import org.apache.sling.event.jobs.jmx.QueuesMBean;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.event.EventConstants;
@@ -80,15 +77,15 @@ import org.slf4j.LoggerFactory;
/**
* Implementation of the job manager.
*/
-@Component(immediate=true)
-@Service(value={JobManager.class, EventHandler.class, Runnable.class})
-@Properties({
- @Property(name="scheduler.period", longValue=60),
- @Property(name="scheduler.concurrent", boolValue=false),
- @Property(name=EventConstants.EVENT_TOPIC,
- value={ResourceHelper.BUNDLE_EVENT_STARTED,
- ResourceHelper.BUNDLE_EVENT_UPDATED})
-})
+@Component(immediate=true,
+ service={JobManager.class, EventHandler.class, Runnable.class},
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation",
+ "scheduler.period:Long=60",
+ "scheduler.concurrent:Boolean=false",
+ EventConstants.EVENT_TOPIC + "=" + ResourceHelper.BUNDLE_EVENT_STARTED,
+ EventConstants.EVENT_TOPIC + "=" + ResourceHelper.BUNDLE_EVENT_UPDATED
+ })
public class JobManagerImpl
implements JobManager, EventHandler, Runnable {
@@ -383,7 +380,7 @@ public class JobManagerImpl
|| type == QueryType.ERROR
|| type == QueryType.GIVEN_UP
|| type == QueryType.STOPPED;
- final List<Job> result = new ArrayList<Job>();
+ final List<Job> result = new ArrayList<>();
final ResourceResolver resolver = this.configuration.createResourceResolver();
final StringBuilder buf = new StringBuilder(64);
try {
@@ -609,7 +606,7 @@ public class JobManagerImpl
final String path = this.configuration.getUniquePath(info.targetId, jobTopic, jobId, jobProperties);
// create properties
- final Map<String, Object> properties = new HashMap<String, Object>();
+ final Map<String, Object> properties = new HashMap<>();
if ( jobProperties != null ) {
for(final Map.Entry<String, Object> entry : jobProperties.entrySet() ) {
@@ -720,7 +717,7 @@ public class JobManagerImpl
errorMessage});
return null;
}
- final List<String> errorList = new ArrayList<String>();
+ final List<String> errorList = new ArrayList<>();
Job result = this.addJobInternal(topic, properties, errorList);
if ( errors != null ) {
errors.addAll(errorList);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/Utility.java b/src/main/java/org/apache/sling/event/impl/jobs/Utility.java
index c93fe51..0449acb 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/Utility.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/Utility.java
@@ -40,8 +40,6 @@ import org.slf4j.Logger;
public abstract class Utility {
- public static volatile boolean LOG_DEPRECATION_WARNINGS = true;
-
/**
* Check if the job topic is a valid OSGI event name (see 113.3.1 of the OSGI spec)
* @return <code>null</code> if the topic is syntactically correct otherwise an error description is returned
@@ -55,7 +53,7 @@ public abstract class Utility {
} catch (final IllegalArgumentException iae) {
message = String.format("Discarding job - job has an illegal job topic '%s'",jobTopic);
}
-
+
} else {
message = "Discarding job - job topic is not of type string";
}
@@ -90,7 +88,7 @@ public abstract class Utility {
* @return New event object.
*/
public static Event toEvent(final Job job) {
- final Map<String, Object> eventProps = new HashMap<String, Object>();
+ final Map<String, Object> eventProps = new HashMap<>();
eventProps.putAll(((JobImpl)job).getProperties());
eventProps.put(ResourceHelper.PROPERTY_JOB_ID, job.getId());
eventProps.remove(JobConsumer.PROPERTY_JOB_ASYNC_HANDLER);
@@ -257,7 +255,7 @@ public abstract class Utility {
* @return Sorted list of children.
*/
public static List<Resource> getSortedChildren(final Logger logger, final String type, final Resource rsrc) {
- final List<Resource> children = new ArrayList<Resource>();
+ final List<Resource> children = new ArrayList<>();
final Iterator<Resource> monthIter = rsrc.listChildren();
while ( monthIter.hasNext() ) {
final Resource monthResource = monthIter.next();
@@ -267,15 +265,4 @@ public abstract class Utility {
Collections.sort(children, RESOURCE_COMPARATOR);
return children;
}
-
- /**
- * Log a deprecation warning on level info into the log
- * @param logger The logger to use
- * @param message The message.
- */
- public static void logDeprecated(final Logger logger, final String message) {
- if ( LOG_DEPRECATION_WARNINGS && logger.isInfoEnabled() ) {
- logger.info("DEPRECATION-WARNING: " + message, new Exception(message));
- }
- }
}
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/ConfigurationConstants.java b/src/main/java/org/apache/sling/event/impl/jobs/config/ConfigurationConstants.java
index 18f4886..d152782 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/ConfigurationConstants.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/ConfigurationConstants.java
@@ -25,14 +25,10 @@ public abstract class ConfigurationConstants {
public static final int NUMBER_OF_PROCESSORS = Runtime.getRuntime().availableProcessors();
- public static final String DEFAULT_TYPE = "UNORDERED";
public static final String DEFAULT_PRIORITY = "NORM";
public static final int DEFAULT_RETRIES = 10;
public static final long DEFAULT_RETRY_DELAY = 2000;
public static final int DEFAULT_MAX_PARALLEL = 15;
- public static final boolean DEFAULT_KEEP_JOBS = false;
- public static final int DEFAULT_THREAD_POOL_SIZE = 0;
- public static final boolean DEFAULT_PREFER_RUN_ON_CREATION_INSTANCE = false;
public static final String PROP_NAME = "queue.name";
public static final String PROP_TYPE = "queue.type";
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfiguration.java b/src/main/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfiguration.java
index c8457a2..4922a72 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfiguration.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfiguration.java
@@ -21,103 +21,108 @@ package org.apache.sling.event.impl.jobs.config;
import java.util.Arrays;
import java.util.Map;
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.ConfigurationPolicy;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.PropertyOption;
-import org.apache.felix.scr.annotations.PropertyUnbounded;
-import org.apache.felix.scr.annotations.Service;
-import org.apache.sling.commons.osgi.PropertiesUtil;
import org.apache.sling.event.impl.support.TopicMatcher;
import org.apache.sling.event.impl.support.TopicMatcherHelper;
import org.apache.sling.event.jobs.QueueConfiguration;
import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.ConfigurationPolicy;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
+import org.osgi.service.metatype.annotations.Option;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@Component(metatype=true,
+@Component(service=InternalQueueConfiguration.class,
name="org.apache.sling.event.jobs.QueueConfiguration",
- label="Apache Sling Job Queue Configuration",
- description="The configuration of a job processing queue.",
- configurationFactory=true, policy=ConfigurationPolicy.REQUIRE)
-@Service(value={InternalQueueConfiguration.class})
-@Properties({
- @Property(name=ConfigurationConstants.PROP_NAME,
- label="Name",
- description="The name of the queue. If matching is used the token {0} can be used to substitute the real value."),
- @Property(name=ConfigurationConstants.PROP_TOPICS,
- unbounded=PropertyUnbounded.ARRAY,
- label="Topics",
+ configurationPolicy=ConfigurationPolicy.REQUIRE,
+ property={
+ "webconsole.configurationFactory.nameHint=Queue: {" + ConfigurationConstants.PROP_NAME + "}",
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ })
+@Designate(ocd = InternalQueueConfiguration.Config.class, factory = true)
+public class InternalQueueConfiguration
+ implements QueueConfiguration, Comparable<InternalQueueConfiguration> {
+
+ @ObjectClassDefinition(name = "Apache Sling Job Queue Configuration",
+ description="The configuration of a job processing queue.")
+ public @interface Config {
+
+ @AttributeDefinition(name="Name",
+ description="The name of the queue. If matching is used the token {0} can be used to substitute the real value.")
+ String queue_name();
+
+ @AttributeDefinition(name="Topics",
description="This value is required and lists the topics processed by "
+ "this queue. The value is a list of strings. If a string ends with a dot, "
+ "all topics in exactly this package match. If the string ends with a star, "
+ "all topics in this package and all subpackages match. If the string neither "
- + "ends with a dot nor with a star, this is assumed to define an exact topic."),
- @Property(name=ConfigurationConstants.PROP_TYPE,
- value=ConfigurationConstants.DEFAULT_TYPE,
- options={@PropertyOption(name="UNORDERED",value="Parallel"),
- @PropertyOption(name="ORDERED",value="Ordered"),
- @PropertyOption(name="TOPIC_ROUND_ROBIN",value="Topic Round Robin")},
- label="Type",
- description="The queue type."),
- @Property(name=ConfigurationConstants.PROP_MAX_PARALLEL,
- doubleValue=ConfigurationConstants.DEFAULT_MAX_PARALLEL,
- label="Maximum Parallel Jobs",
- description="The maximum number of parallel jobs started for this queue. "
- + "A value of -1 is substituted with the number of available processors. "
- + "Positive integer values specify number of processors to use. Can be greater than number of processors. "
- + "A decimal number between 0.0 and 1.0 is treated as a fraction of available processors. "
- + "For example 0.5 means half of the available processors. For ordered queue types this value is ignored (always enforced to be 1)."),
- @Property(name=ConfigurationConstants.PROP_RETRIES,
- intValue=ConfigurationConstants.DEFAULT_RETRIES,
- label="Maximum Retries",
- description="The maximum number of times a failed job slated "
- + "for retries is actually retried. If a job has been retried this number of "
- + "times and still fails, it is not rescheduled and assumed to have failed. The "
- + "default value is 10."),
- @Property(name=ConfigurationConstants.PROP_RETRY_DELAY,
- longValue=ConfigurationConstants.DEFAULT_RETRY_DELAY,
- label="Retry Delay",
- description="The number of milliseconds to sleep between two "
- + "consecutive retries of a job which failed and was set to be retried. The "
- + "default value is 2 seconds. This value is only relevant if there is a single "
- + "failed job in the queue. If there are multiple failed jobs, each job is "
- + "retried in turn without an intervening delay."),
- @Property(name=ConfigurationConstants.PROP_PRIORITY,
- value=ConfigurationConstants.DEFAULT_PRIORITY,
- options={@PropertyOption(name="NORM",value="Norm"),
- @PropertyOption(name="MIN",value="Min"),
- @PropertyOption(name="MAX",value="Max")},
- label="Priority",
- description="The priority for the threads used by this queue. Default is norm."),
- @Property(name=ConfigurationConstants.PROP_KEEP_JOBS,
- boolValue=ConfigurationConstants.DEFAULT_KEEP_JOBS,
- label="Keep History",
+ + "ends with a dot nor with a star, this is assumed to define an exact topic.")
+ String[] queue_topics();
+
+ @AttributeDefinition(name = "Type",
+ description="The queue type.",
+ options = {@Option(label="UNORDERED",value="Parallel"),
+ @Option(label="ORDERED",value="Ordered"),
+ @Option(label="TOPIC_ROUND_ROBIN",value="Topic Round Robin")})
+ String queue_type() default "UNORDERED";
+
+ @AttributeDefinition(
+ name="Priority",
+ description="The priority for the threads used by this queue. Default is norm.",
+ options = {
+ @Option(label="NORM",value="Norm"),
+ @Option(label="MIN",value="Min"),
+ @Option(label="MAX",value="Max")
+ })
+ String queue_priority() default ConfigurationConstants.DEFAULT_PRIORITY;
+
+ @AttributeDefinition(name="Maximum Retries",
+ description="The maximum number of times a failed job slated "
+ + "for retries is actually retried. If a job has been retried this number of "
+ + "times and still fails, it is not rescheduled and assumed to have failed. The "
+ + "default value is 10.")
+ int queue_retries() default ConfigurationConstants.DEFAULT_RETRIES;
+
+ @AttributeDefinition(name="Retry Delay",
+ description="The number of milliseconds to sleep between two "
+ + "consecutive retries of a job which failed and was set to be retried. The "
+ + "default value is 2 seconds. This value is only relevant if there is a single "
+ + "failed job in the queue. If there are multiple failed jobs, each job is "
+ + "retried in turn without an intervening delay.")
+ long queue_retrydelay() default ConfigurationConstants.DEFAULT_RETRY_DELAY;
+
+ @AttributeDefinition(name="Maximum Parallel Jobs",
+ description="The maximum number of parallel jobs started for this queue. "
+ + "A value of -1 is substituted with the number of available processors. "
+ + "Positive integer values specify number of processors to use. Can be greater than number of processors. "
+ + "A decimal number between 0.0 and 1.0 is treated as a fraction of available processors. "
+ + "For example 0.5 means half of the available processors. For ordered queue types this value is ignored (always enforced to be 1).")
+ double queue_maxparallel() default ConfigurationConstants.DEFAULT_MAX_PARALLEL;
+
+ @AttributeDefinition(name="Keep History",
description="If this option is enabled, successful finished jobs are kept "
- + "to provide a complete history."),
- @Property(name=ConfigurationConstants.PROP_PREFER_RUN_ON_CREATION_INSTANCE,
- boolValue=ConfigurationConstants.DEFAULT_PREFER_RUN_ON_CREATION_INSTANCE,
- label="Prefer Creation Instance",
+ + "to provide a complete history.")
+ boolean queue_keepJobs() default false;
+
+ @AttributeDefinition(name="Prefer Creation Instance",
description="If this option is enabled, the jobs are tried to "
- + "be run on the instance where the job was created."),
- @Property(name=ConfigurationConstants.PROP_THREAD_POOL_SIZE,
- intValue=ConfigurationConstants.DEFAULT_THREAD_POOL_SIZE,
- label="Thread Pool Size",
- description="Optional configuration value for a thread pool to be used by "
- + "this queue. If this is value has a positive number of threads configuration, this queue uses "
- + "an own thread pool with the configured number of threads."),
- @Property(name=Constants.SERVICE_RANKING,
- intValue=0,
- propertyPrivate=false,
- label="Ranking",
+ + "be run on the instance where the job was created.")
+ boolean queue_preferRunOnCreationInstance() default false;
+
+ @AttributeDefinition(name="Thread Pool Size",
+ description="Optional configuration value for a thread pool to be used by "
+ + "this queue. If this is value has a positive number of threads configuration, this queue uses "
+ + "an own thread pool with the configured number of threads.")
+ int queue_threadPoolSize() default 0;
+
+ @AttributeDefinition(name="Ranking",
description="Integer value defining the ranking of this queue configuration. "
- + "If more than one queue matches a job topic, the one with the highest ranking is used."),
- @Property(name="webconsole.configurationFactory.nameHint", value="Queue: {" + ConfigurationConstants.PROP_NAME + "}")
-})
-public class InternalQueueConfiguration
- implements QueueConfiguration, Comparable<InternalQueueConfiguration> {
+ + "If more than one queue matches a job topic, the one with the highest ranking is used.")
+ int service_ranking() default 0;
+ }
/** Logger. */
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -166,9 +171,9 @@ public class InternalQueueConfiguration
/**
* Create a new configuration from a config
*/
- public static InternalQueueConfiguration fromConfiguration(final Map<String, Object> params) {
+ public static InternalQueueConfiguration fromConfiguration(final Map<String, Object> props, final Config config) {
final InternalQueueConfiguration c = new InternalQueueConfiguration();
- c.activate(params);
+ c.activate(props, config);
return c;
}
@@ -180,28 +185,27 @@ public class InternalQueueConfiguration
* Create a new queue configuration
*/
@Activate
- protected void activate(final Map<String, Object> params) {
- this.name = PropertiesUtil.toString(params.get(ConfigurationConstants.PROP_NAME), null);
+ protected void activate(final Map<String, Object> props, final Config config) {
+ this.name = config.queue_name();
try {
- this.priority = ThreadPriority.valueOf(PropertiesUtil.toString(params.get(ConfigurationConstants.PROP_PRIORITY), ConfigurationConstants.DEFAULT_PRIORITY));
+ this.priority = ThreadPriority.valueOf(config.queue_priority());
} catch ( final IllegalArgumentException iae) {
- logger.warn("Invalid value for queue priority. Using default instead of : {}", params.get(ConfigurationConstants.PROP_PRIORITY));
+ logger.warn("Invalid value for queue priority. Using default instead of : {}", config.queue_priority());
this.priority = ThreadPriority.valueOf(ConfigurationConstants.DEFAULT_PRIORITY);
}
try {
- this.type = Type.valueOf(PropertiesUtil.toString(params.get(ConfigurationConstants.PROP_TYPE), ConfigurationConstants.DEFAULT_TYPE));
+ this.type = Type.valueOf(config.queue_type());
} catch ( final IllegalArgumentException iae) {
- logger.error("Invalid value for queue type configuration: {}", params.get(ConfigurationConstants.PROP_TYPE));
+ logger.error("Invalid value for queue type configuration: {}", config.queue_type());
this.type = null;
}
- this.retries = PropertiesUtil.toInteger(params.get(ConfigurationConstants.PROP_RETRIES), ConfigurationConstants.DEFAULT_RETRIES);
- this.retryDelay = PropertiesUtil.toLong(params.get(ConfigurationConstants.PROP_RETRY_DELAY), ConfigurationConstants.DEFAULT_RETRY_DELAY);
+ this.retries = config.queue_retries();
+ this.retryDelay = config.queue_retrydelay();
// Float values are treated as percentage. int values are treated as number of cores, -1 == all available
// Note: the value is based on the core count at startup. It will not change dynamically if core count changes.
int cores = ConfigurationConstants.NUMBER_OF_PROCESSORS;
- final double inMaxParallel = PropertiesUtil.toDouble(params.get(ConfigurationConstants.PROP_MAX_PARALLEL),
- ConfigurationConstants.DEFAULT_MAX_PARALLEL);
+ final double inMaxParallel = config.queue_maxparallel();
logger.debug("Max parallel for queue {} is {}", this.name, inMaxParallel);
if ((inMaxParallel == Math.floor(inMaxParallel)) && !Double.isInfinite(inMaxParallel)) {
// integral type
@@ -224,18 +228,18 @@ public class InternalQueueConfiguration
if ( this.type == Type.ORDERED ) {
this.maxParallelProcesses = 1;
}
- final String[] topicsParam = PropertiesUtil.toStringArray(params.get(ConfigurationConstants.PROP_TOPICS));
+ final String[] topicsParam = config.queue_topics();
this.matchers = TopicMatcherHelper.buildMatchers(topicsParam);
if ( this.matchers == null ) {
this.topics = null;
} else {
this.topics = topicsParam;
}
- this.keepJobs = PropertiesUtil.toBoolean(params.get(ConfigurationConstants.PROP_KEEP_JOBS), ConfigurationConstants.DEFAULT_KEEP_JOBS);
- this.serviceRanking = PropertiesUtil.toInteger(params.get(Constants.SERVICE_RANKING), 0);
- this.ownThreadPoolSize = PropertiesUtil.toInteger(params.get(ConfigurationConstants.PROP_THREAD_POOL_SIZE), ConfigurationConstants.DEFAULT_THREAD_POOL_SIZE);
- this.preferCreationInstance = PropertiesUtil.toBoolean(params.get(ConfigurationConstants.PROP_PREFER_RUN_ON_CREATION_INSTANCE), ConfigurationConstants.DEFAULT_PREFER_RUN_ON_CREATION_INSTANCE);
- this.pid = (String)params.get(Constants.SERVICE_PID);
+ this.keepJobs = config.queue_keepJobs();
+ this.serviceRanking = config.service_ranking();
+ this.ownThreadPoolSize = config.queue_threadPoolSize();
+ this.preferCreationInstance = config.queue_preferRunOnCreationInstance();
+ this.pid = (String)props.get(Constants.SERVICE_PID);
this.valid = this.checkIsValid();
}
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java b/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
index 3cc8b72..4f75a4c 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
@@ -27,14 +27,6 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
-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.Modified;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.api.resource.ResourceResolver;
@@ -46,7 +38,6 @@ import org.apache.sling.discovery.TopologyEvent.Type;
import org.apache.sling.discovery.TopologyEventListener;
import org.apache.sling.discovery.commons.InitDelayingTopologyEventListener;
import org.apache.sling.event.impl.EnvironmentComponent;
-import org.apache.sling.event.impl.jobs.Utility;
import org.apache.sling.event.impl.jobs.tasks.CheckTopologyTask;
import org.apache.sling.event.impl.jobs.tasks.FindUnfinishedJobsTask;
import org.apache.sling.event.impl.jobs.tasks.UpgradeTask;
@@ -54,6 +45,15 @@ import org.apache.sling.event.impl.support.Environment;
import org.apache.sling.event.impl.support.ResourceHelper;
import org.apache.sling.event.jobs.Job;
import org.apache.sling.serviceusermapping.ServiceUserMapped;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,35 +61,32 @@ import org.slf4j.LoggerFactory;
* Configuration of the job handling
*
*/
-@Component(immediate=true, metatype=true,
- label="Apache Sling Job Manager",
- description="This is the central service of the job handling.",
- name="org.apache.sling.event.impl.jobs.jcr.PersistenceHandler")
-@Service(value={JobManagerConfiguration.class})
-@Properties({
- @Property(name=JobManagerConfiguration.PROPERTY_DISABLE_DISTRIBUTION,
- boolValue=JobManagerConfiguration.DEFAULT_DISABLE_DISTRIBUTION,
- label="Disable Distribution",
- description="If the distribution is disabled, all jobs will be processed on the leader only! "
- + "Please use this switch with care."),
- @Property(name=JobManagerConfiguration.PROPERTY_LOG_DEPRECATION_WARNINGS,
- boolValue=JobManagerConfiguration.DEFAULT_LOG_DEPRECATION_WARNINGS,
- label="Deprecation Warnings",
- description="If this switch is enabled, deprecation warnings will be logged with the INFO level."),
- @Property(name=JobManagerConfiguration.PROPERTY_STARTUP_DELAY,
- longValue=JobManagerConfiguration.DEFAULT_STARTUP_DELAY,
- label="Startup Delay",
- description="Specify amount in seconds that job manager waits on startup before starting with job handling. "
- + "This can be used to allow enough time to restart a cluster before jobs are eventually reassigned."),
- @Property(name=JobManagerConfiguration.PROPERTY_REPOSITORY_PATH,
- value=JobManagerConfiguration.DEFAULT_REPOSITORY_PATH, propertyPrivate=true),
- @Property(name=JobManagerConfiguration.PROPERTY_SCHEDULED_JOBS_PATH,
- value=JobManagerConfiguration.DEFAULT_SCHEDULED_JOBS_PATH, propertyPrivate=true),
- @Property(name=JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY,
- longValue=JobManagerConfiguration.DEFAULT_BACKGROUND_LOAD_DELAY, propertyPrivate=true),
+@Component(immediate=true,
+ service=JobManagerConfiguration.class,
+ name="org.apache.sling.event.impl.jobs.jcr.PersistenceHandler",
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation",
+ JobManagerConfiguration.PROPERTY_REPOSITORY_PATH + "=" + JobManagerConfiguration.DEFAULT_REPOSITORY_PATH,
+ JobManagerConfiguration.PROPERTY_SCHEDULED_JOBS_PATH + "=" + JobManagerConfiguration.DEFAULT_SCHEDULED_JOBS_PATH,
+ JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY + ":Long=" + JobManagerConfiguration.DEFAULT_BACKGROUND_LOAD_DELAY
})
+@Designate(ocd = JobManagerConfiguration.Config.class)
public class JobManagerConfiguration {
+ @ObjectClassDefinition(name = "Apache Sling Job Manager",
+ description="This is the central service of the job handling.")
+ public @interface Config {
+
+ @AttributeDefinition(name = "Disable Distribution",
+ description="If the distribution is disabled, all jobs will be processed on the leader only! "
+ + "Please use this switch with care.")
+ boolean job_consumermanager_disableDistribution() default false;
+
+ @AttributeDefinition(name = "Startup Delay",
+ description="Specify amount in seconds that job manager waits on startup before starting with job handling. "
+ + "This can be used to allow enough time to restart a cluster before jobs are eventually reassigned.")
+ long startup_delay() default 30;
+ }
/** Logger. */
private final Logger logger = LoggerFactory.getLogger("org.apache.sling.event.impl.jobs");
@@ -102,12 +99,6 @@ public class JobManagerConfiguration {
/** Default background load delay. */
public static final long DEFAULT_BACKGROUND_LOAD_DELAY = 10;
- /** Default startup delay. */
- public static final long DEFAULT_STARTUP_DELAY = 30;
-
- /** Default for disabling the distribution. */
- public static final boolean DEFAULT_DISABLE_DISTRIBUTION = false;
-
/** Default resource path for scheduled jobs. */
public static final String DEFAULT_SCHEDULED_JOBS_PATH = "/var/eventing/scheduled-jobs";
@@ -117,24 +108,9 @@ public class JobManagerConfiguration {
/** The background loader waits this time of seconds after startup before loading events from the repository. (in secs) */
public static final String PROPERTY_BACKGROUND_LOAD_DELAY = "load.delay";
- /** The entire job handling waits time amount of seconds until it starts - to allow avoiding reassign on restart of a cluster */
- public static final String PROPERTY_STARTUP_DELAY = "startup.delay";
-
- /** Configuration switch for distributing the jobs. */
- public static final String PROPERTY_DISABLE_DISTRIBUTION = "job.consumermanager.disableDistribution";
-
/** Configuration property for the scheduled jobs path. */
public static final String PROPERTY_SCHEDULED_JOBS_PATH = "job.scheduled.jobs.path";
- /** Default value for background loading. */
- public static final boolean DEFAULT_BACKGROUND_LOAD_SEARCH = true;
-
- /** Configuration property for deprecation warnings. */
- public static final String PROPERTY_LOG_DEPRECATION_WARNINGS = "job.log.deprecation";
-
- /** Default value for deprecation warnings. */
- public static final boolean DEFAULT_LOG_DEPRECATION_WARNINGS = true;
-
/** The jobs base path with a slash. */
private String jobsBasePathWithSlash;
@@ -173,7 +149,7 @@ public class JobManagerConfiguration {
private String scheduledJobsPathWithSlash;
/** List of topology awares. */
- private final List<ConfigurationChangeListener> listeners = new ArrayList<ConfigurationChangeListener>();
+ private final List<ConfigurationChangeListener> listeners = new ArrayList<>();
/** The environment component. */
@Reference
@@ -200,11 +176,12 @@ public class JobManagerConfiguration {
/**
* Activate this component.
* @param props Configuration properties
+ * @param config Configuration properties
* @throws RuntimeException If the default paths can't be created
*/
@Activate
- protected void activate(final Map<String, Object> props) {
- this.update(props);
+ protected void activate(final Map<String, Object> props, final Config config) {
+ this.update(props, config);
this.jobsBasePathWithSlash = PropertiesUtil.toString(props.get(PROPERTY_REPOSITORY_PATH),
DEFAULT_REPOSITORY_PATH) + '/';
@@ -240,7 +217,7 @@ public class JobManagerConfiguration {
// SLING-5560 : use an InitDelayingTopologyEventListener
if (this.startupDelay > 0) {
- logger.debug("activate: job manager will start in {} sec. ({})", this.startupDelay, PROPERTY_STARTUP_DELAY);
+ logger.debug("activate: job manager will start in {} sec. ({})", this.startupDelay, config.startup_delay());
this.startupDelayListener = new InitDelayingTopologyEventListener(startupDelay, new TopologyEventListener() {
@Override
@@ -249,7 +226,7 @@ public class JobManagerConfiguration {
}
}, this.scheduler, logger);
} else {
- logger.debug("activate: job manager will start without delay. ({}:{})", PROPERTY_STARTUP_DELAY, this.startupDelay);
+ logger.debug("activate: job manager will start without delay. ({}:{})", config.startup_delay(), this.startupDelay);
}
}
@@ -257,14 +234,13 @@ public class JobManagerConfiguration {
* Update with a new configuration
*/
@Modified
- protected void update(final Map<String, Object> props) {
- this.disabledDistribution = PropertiesUtil.toBoolean(props.get(PROPERTY_DISABLE_DISTRIBUTION), DEFAULT_DISABLE_DISTRIBUTION);
+ protected void update(final Map<String, Object> props, final Config config) {
+ this.disabledDistribution = config.job_consumermanager_disableDistribution();
this.backgroundLoadDelay = PropertiesUtil.toLong(props.get(PROPERTY_BACKGROUND_LOAD_DELAY), DEFAULT_BACKGROUND_LOAD_DELAY);
// SLING-5560: note that currently you can't change the startupDelay to have
// an immediate effect - it will only have an effect on next activation.
// (as 'startup delay runnable' is already scheduled in activate)
- this.startupDelay = PropertiesUtil.toLong(props.get(PROPERTY_STARTUP_DELAY), DEFAULT_STARTUP_DELAY);
- Utility.LOG_DEPRECATION_WARNINGS = PropertiesUtil.toBoolean(props.get(PROPERTY_LOG_DEPRECATION_WARNINGS), DEFAULT_LOG_DEPRECATION_WARNINGS);
+ this.startupDelay = config.startup_delay();
}
/**
@@ -622,7 +598,7 @@ public class JobManagerConfiguration {
}
}
- private final Map<String, Job> retryList = new HashMap<String, Job>();
+ private final Map<String, Job> retryList = new HashMap<>();
public void addJobToRetryList(final Job job) {
synchronized ( retryList ) {
@@ -631,7 +607,7 @@ public class JobManagerConfiguration {
}
public List<Job> clearJobRetryList() {
- final List<Job> result = new ArrayList<Job>();
+ final List<Job> result = new ArrayList<>();
synchronized ( this.retryList ) {
result.addAll(retryList.values());
retryList.clear();
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/MainQueueConfiguration.java b/src/main/java/org/apache/sling/event/impl/jobs/config/MainQueueConfiguration.java
index 9287e6b..9f45b83 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/MainQueueConfiguration.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/MainQueueConfiguration.java
@@ -18,17 +18,17 @@
*/
package org.apache.sling.event.impl.jobs.config;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Modified;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.PropertyOption;
-import org.apache.felix.scr.annotations.Service;
+import java.lang.annotation.Annotation;
+import java.util.Collections;
+
import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
+import org.osgi.service.metatype.annotations.Option;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,42 +37,49 @@ import org.slf4j.LoggerFactory;
* This is the configuration for the main queue.
*
*/
-@Component(label="Apache Sling Job Default Queue",
- description="The configuration of the default job queue.",
- name="org.apache.sling.event.impl.jobs.DefaultJobManager",
- metatype=true)
-@Service(value=MainQueueConfiguration.class)
-@Properties({
- @Property(name=ConfigurationConstants.PROP_PRIORITY,
- value=ConfigurationConstants.DEFAULT_PRIORITY,
- options={@PropertyOption(name="NORM",value="Norm"),
- @PropertyOption(name="MIN",value="Min"),
- @PropertyOption(name="MAX",value="Max")},
- label="Priority",
- description="The priority for the threads used by this queue. Default is norm."),
- @Property(name=ConfigurationConstants.PROP_RETRIES,
- intValue=ConfigurationConstants.DEFAULT_RETRIES,
- label="Maximum Retries",
- description="The maximum number of times a failed job slated "
- + "for retries is actually retried. If a job has been retried this number of "
- + "times and still fails, it is not rescheduled and assumed to have failed. The "
- + "default value is 10."),
- @Property(name=ConfigurationConstants.PROP_RETRY_DELAY,
- longValue=ConfigurationConstants.DEFAULT_RETRY_DELAY,
- label="Retry Delay",
- description="The number of milliseconds to sleep between two "
- + "consecutive retries of a job which failed and was set to be retried. The "
- + "default value is 2 seconds. This value is only relevant if there is a single "
- + "failed job in the queue. If there are multiple failed jobs, each job is "
- + "retried in turn without an intervening delay."),
- @Property(name=ConfigurationConstants.PROP_MAX_PARALLEL,
- intValue=ConfigurationConstants.DEFAULT_MAX_PARALLEL,
- label="Maximum Parallel Jobs",
- description="The maximum number of parallel jobs started for this queue. "
- + "A value of -1 is substituted with the number of available processors."),
-})
+@Component(name="org.apache.sling.event.impl.jobs.DefaultJobManager",
+ service=MainQueueConfiguration.class,
+ property={
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ })
+@Designate(ocd=MainQueueConfiguration.Config.class)
public class MainQueueConfiguration {
+ @ObjectClassDefinition(name = "Apache Sling Job Default Queue",
+ description="The configuration of the default job queue.")
+ public @interface Config {
+
+ @AttributeDefinition(
+ name="Priority",
+ description="The priority for the threads used by this queue. Default is norm.",
+ options = {
+ @Option(label="NORM",value="Norm"),
+ @Option(label="MIN",value="Min"),
+ @Option(label="MAX",value="Max")
+ })
+ String queue_priority() default ConfigurationConstants.DEFAULT_PRIORITY;
+
+ @AttributeDefinition(name="Maximum Retries",
+ description="The maximum number of times a failed job slated "
+ + "for retries is actually retried. If a job has been retried this number of "
+ + "times and still fails, it is not rescheduled and assumed to have failed. The "
+ + "default value is 10.")
+ int queue_retries() default ConfigurationConstants.DEFAULT_RETRIES;
+
+ @AttributeDefinition(name="Retry Delay",
+ description="The number of milliseconds to sleep between two "
+ + "consecutive retries of a job which failed and was set to be retried. The "
+ + "default value is 2 seconds. This value is only relevant if there is a single "
+ + "failed job in the queue. If there are multiple failed jobs, each job is "
+ + "retried in turn without an intervening delay.")
+ long queue_retrydelay() default ConfigurationConstants.DEFAULT_RETRY_DELAY;
+
+ @AttributeDefinition(name="Maximum Parallel Jobs",
+ description="The maximum number of parallel jobs started for this queue. "
+ + "A value of -1 is substituted with the number of available processors.")
+ int queue_maxparallel() default ConfigurationConstants.DEFAULT_MAX_PARALLEL;
+ }
+
public static final String MAIN_QUEUE_NAME = "<main queue>";
/** Default logger. */
@@ -82,27 +89,84 @@ public class MainQueueConfiguration {
/**
* Activate this component.
- * @param props Configuration properties
+ * @param config Configuration properties
*/
@Activate
- protected void activate(final Map<String, Object> props) {
- this.update(props);
+ protected void activate(final Config config) {
+ this.update(config);
}
/**
* Configure this component.
- * @param props Configuration properties
+ * @param config Configuration properties
*/
@Modified
- protected void update(final Map<String, Object> props) {
- // create a new dictionary with the missing info and do some sanity puts
- final Map<String, Object> queueProps = new HashMap<String, Object>(props);
- queueProps.put(ConfigurationConstants.PROP_TOPICS, "*");
- queueProps.put(ConfigurationConstants.PROP_NAME, MAIN_QUEUE_NAME);
- queueProps.put(ConfigurationConstants.PROP_TYPE, InternalQueueConfiguration.Type.UNORDERED);
- queueProps.put(Constants.SERVICE_PID, "org.apache.sling.event.impl.jobs.DefaultJobManager");
- logger.debug("properties for queue {}: {}", MAIN_QUEUE_NAME, queueProps);
- this.mainConfiguration = InternalQueueConfiguration.fromConfiguration(queueProps);
+ protected void update(final Config config) {
+ logger.debug("properties for queue {}: {}", MAIN_QUEUE_NAME, config);
+ this.mainConfiguration = InternalQueueConfiguration.fromConfiguration(
+ Collections.singletonMap(Constants.SERVICE_PID, (Object)"org.apache.sling.event.impl.jobs.DefaultJobManager"),
+ new InternalQueueConfiguration.Config() {
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return InternalQueueConfiguration.Config.class;
+ }
+
+ @Override
+ public int service_ranking() {
+ return 0;
+ }
+
+ @Override
+ public String queue_type() {
+ return InternalQueueConfiguration.Type.UNORDERED.name();
+ }
+
+ @Override
+ public String[] queue_topics() {
+ return new String[] {"*"};
+ }
+
+ @Override
+ public int queue_threadPoolSize() {
+ return 0;
+ }
+
+ @Override
+ public long queue_retrydelay() {
+ return config.queue_retrydelay();
+ }
+
+ @Override
+ public int queue_retries() {
+ return config.queue_retries();
+ }
+
+ @Override
+ public String queue_priority() {
+ return config.queue_priority();
+ }
+
+ @Override
+ public boolean queue_preferRunOnCreationInstance() {
+ return false;
+ }
+
+ @Override
+ public String queue_name() {
+ return MAIN_QUEUE_NAME;
+ }
+
+ @Override
+ public double queue_maxparallel() {
+ return config.queue_maxparallel();
+ }
+
+ @Override
+ public boolean queue_keepJobs() {
+ return false;
+ }
+ });
}
/**
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/QueueConfigurationManager.java b/src/main/java/org/apache/sling/event/impl/jobs/config/QueueConfigurationManager.java
index ce2b79a..753568b 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/QueueConfigurationManager.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/QueueConfigurationManager.java
@@ -22,22 +22,21 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import org.apache.felix.scr.annotations.Component;
-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.event.impl.support.ResourceHelper;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.osgi.service.component.annotations.ReferencePolicy;
/**
* The queue manager manages queue configurations.
*/
-@Component
-@Service(value=QueueConfigurationManager.class)
-@Reference(referenceInterface=InternalQueueConfiguration.class, policy=ReferencePolicy.DYNAMIC,
- cardinality=ReferenceCardinality.OPTIONAL_MULTIPLE,
- bind="bindConfig", unbind="unbindConfig", updated="updateConfig")
+@Component(service = QueueConfigurationManager.class,
+property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
public class QueueConfigurationManager {
/** Empty configuration array. */
@@ -47,7 +46,7 @@ public class QueueConfigurationManager {
private volatile InternalQueueConfiguration[] orderedConfigs = EMPTY_CONFIGS;
/** All configurations. */
- private final List<InternalQueueConfiguration> configurations = new ArrayList<InternalQueueConfiguration>();
+ private final List<InternalQueueConfiguration> configurations = new ArrayList<>();
/** The main queue configuration. */
@Reference
@@ -57,6 +56,9 @@ public class QueueConfigurationManager {
* Add a new queue configuration.
* @param config A new queue configuration.
*/
+ @Reference(service=InternalQueueConfiguration.class, policy=ReferencePolicy.DYNAMIC,
+ cardinality=ReferenceCardinality.MULTIPLE,
+ updated="updateConfig")
protected void bindConfig(final InternalQueueConfiguration config) {
synchronized ( configurations ) {
configurations.add(config);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyHandler.java b/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyHandler.java
index 2c49ca2..58da224 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyHandler.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyHandler.java
@@ -22,13 +22,13 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
-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.Service;
import org.apache.sling.discovery.TopologyEvent;
import org.apache.sling.discovery.TopologyEventListener;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,8 +36,10 @@ import org.slf4j.LoggerFactory;
* This topology handler is handling the topology change events asynchronously
* and processes them by queuing them.
*/
-@Component
-@Service(value = TopologyEventListener.class)
+@Component(service = TopologyEventListener.class,
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ })
public class TopologyHandler implements TopologyEventListener, Runnable {
/** The logger. */
@@ -47,7 +49,7 @@ public class TopologyHandler implements TopologyEventListener, Runnable {
private JobManagerConfiguration configuration;
/** A local queue for async handling of the events */
- private final BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<QueueItem>();
+ private final BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>();
/** Active flag. */
private final AtomicBoolean isActive = new AtomicBoolean(false);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/console/InventoryPlugin.java b/src/main/java/org/apache/sling/event/impl/jobs/console/InventoryPlugin.java
index 619136c..f3f3a18 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/console/InventoryPlugin.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/console/InventoryPlugin.java
@@ -31,11 +31,6 @@ import java.util.Map;
import org.apache.felix.inventory.Format;
import org.apache.felix.inventory.InventoryPrinter;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.event.impl.jobs.JobConsumerManager;
import org.apache.sling.event.impl.jobs.config.InternalQueueConfiguration;
@@ -48,19 +43,23 @@ import org.apache.sling.event.jobs.ScheduleInfo;
import org.apache.sling.event.jobs.ScheduledJobInfo;
import org.apache.sling.event.jobs.Statistics;
import org.apache.sling.event.jobs.TopicStatistics;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
/**
* This is a inventory plugin displaying the active queues, some statistics
* and the configurations.
* @since 3.2
*/
-@Component
-@Service(value={InventoryPrinter.class})
-@Properties({
- @Property(name=InventoryPrinter.NAME, value="slingjobs"),
- @Property(name=InventoryPrinter.TITLE, value="Sling Jobs"),
- @Property(name=InventoryPrinter.FORMAT, value={"TEXT", "JSON"}),
- @Property(name=InventoryPrinter.WEBCONSOLE, boolValue=false)
+@Component(service={InventoryPrinter.class},
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation",
+ InventoryPrinter.NAME + "=slingjobs",
+ InventoryPrinter.TITLE + "=Sling Jobs",
+ InventoryPrinter.FORMAT + "=TEXT",
+ InventoryPrinter.FORMAT + "=JSON",
+ InventoryPrinter.WEBCONSOLE + ":Boolean=false"
})
public class InventoryPlugin implements InventoryPrinter {
@@ -325,7 +324,7 @@ public class InventoryPlugin implements InventoryPrinter {
final Iterator<Map.Entry<String, List<InstanceDescription>>> iter = instanceCaps.entrySet().iterator();
while ( iter.hasNext() ) {
final Map.Entry<String, List<InstanceDescription>> entry = iter.next();
- final List<String> instances = new ArrayList<String>();
+ final List<String> instances = new ArrayList<>();
for(final InstanceDescription id : entry.getValue()) {
if ( id.isLocal() ) {
instances.add("local");
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/console/WebConsolePlugin.java b/src/main/java/org/apache/sling/event/impl/jobs/console/WebConsolePlugin.java
index ffe94b5..00b08ad 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/console/WebConsolePlugin.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/console/WebConsolePlugin.java
@@ -33,11 +33,6 @@ import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.api.request.ResponseUtil;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.event.impl.jobs.JobConsumerManager;
@@ -53,6 +48,9 @@ import org.apache.sling.event.jobs.ScheduledJobInfo;
import org.apache.sling.event.jobs.Statistics;
import org.apache.sling.event.jobs.TopicStatistics;
import org.apache.sling.event.jobs.consumer.JobConsumer;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,13 +59,13 @@ import org.slf4j.LoggerFactory;
* and the configurations.
* @since 3.0
*/
-@Component
-@Service(value={javax.servlet.Servlet.class, JobConsumer.class})
-@Properties({
- @Property(name="felix.webconsole.label", value="slingevent"),
- @Property(name="felix.webconsole.title", value="Jobs"),
- @Property(name="felix.webconsole.category", value="Sling"),
- @Property(name=JobConsumer.PROPERTY_TOPICS, value={"sling/webconsole/test"})
+@Component(service={javax.servlet.Servlet.class, JobConsumer.class},
+property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation",
+ "felix.webconsole.label=slingevent",
+ "felix.webconsole.title=Jobs",
+ "felix.webconsole.category=SLING",
+ JobConsumer.PROPERTY_TOPICS + "=sling/webconsole/test"
})
public class WebConsolePlugin extends HttpServlet implements JobConsumer {
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/jmx/AllJobStatisticsMBean.java b/src/main/java/org/apache/sling/event/impl/jobs/jmx/AllJobStatisticsMBean.java
index 0e696d6..154a926 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/jmx/AllJobStatisticsMBean.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/jmx/AllJobStatisticsMBean.java
@@ -17,23 +17,24 @@
*/
package org.apache.sling.event.impl.jobs.jmx;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.event.jobs.JobManager;
import org.apache.sling.event.jobs.Statistics;
import org.apache.sling.event.jobs.jmx.StatisticsMBean;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
-@Component
-@Service(value = StatisticsMBean.class)
-@Properties(@Property(name = "jmx.objectname", value = "org.apache.sling:type=queues,name=AllQueues"))
+@Component(service = StatisticsMBean.class,
+ property = {
+ "jmx.objectname=org.apache.sling:type=queues,name=AllQueues",
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
public class AllJobStatisticsMBean extends AbstractJobStatistics {
private static final long TTL = 1000L;
private long agregateStatisticsTTL = 0L;
private Statistics aggregateStatistics;
+
@Reference
private JobManager jobManager;
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/jmx/QueuesMBeanImpl.java b/src/main/java/org/apache/sling/event/impl/jobs/jmx/QueuesMBeanImpl.java
index 83a56e9..3227266 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/jmx/QueuesMBeanImpl.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/jmx/QueuesMBeanImpl.java
@@ -32,26 +32,26 @@ import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.StandardEmitterMBean;
-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.Property;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.event.jobs.Queue;
import org.apache.sling.event.jobs.jmx.QueuesMBean;
import org.apache.sling.event.jobs.jmx.StatisticsMBean;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
-
-@Component
-@Service(value = { QueuesMBean.class })
-@Property(name = "jmx.objectname", value = "org.apache.sling:type=queues,name=QueueNames")
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+
+@Component(service = QueuesMBean.class,
+property = {
+ "jmx.objectname=org.apache.sling:type=queues,name=QueueNames",
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
public class QueuesMBeanImpl extends StandardEmitterMBean implements QueuesMBean {
private static final String QUEUE_NOTIFICATION = "org.apache.sling.event.queue";
private static final String[] NOTIFICATION_TYPES = { QUEUE_NOTIFICATION };
- private Map<String, QueueMBeanHolder> queues = new ConcurrentHashMap<String, QueueMBeanHolder>();
+ private Map<String, QueueMBeanHolder> queues = new ConcurrentHashMap<>();
private String[] names;
private AtomicLong sequence = new AtomicLong(System.currentTimeMillis());
private BundleContext bundleContext;
@@ -147,7 +147,7 @@ public class QueuesMBeanImpl extends StandardEmitterMBean implements QueuesMBean
private QueueMBeanHolder add(Queue queue) {
QueueMBeanImpl queueMBean = new QueueMBeanImpl(queue);
- ServiceRegistration serviceRegistration = bundleContext
+ ServiceRegistration<?> serviceRegistration = bundleContext
.registerService(StatisticsMBean.class.getName(), queueMBean,
createProperties(
"jmx.objectname","org.apache.sling:type=queues,name="+queue.getName(),
@@ -160,7 +160,7 @@ public class QueuesMBeanImpl extends StandardEmitterMBean implements QueuesMBean
}
private Dictionary<String, Object> createProperties(Object ... values) {
- Dictionary<String, Object> props = new Hashtable<String, Object>();
+ Dictionary<String, Object> props = new Hashtable<>();
for ( int i = 0; i < values.length; i+=2) {
props.put((String) values[i], values[i+1]);
}
@@ -175,7 +175,7 @@ public class QueuesMBeanImpl extends StandardEmitterMBean implements QueuesMBean
@Override
public String[] getQueueNames() {
if (names == null) {
- List<String> lnames = new ArrayList<String>(queues.keySet());
+ List<String> lnames = new ArrayList<>(queues.keySet());
Collections.sort(lnames);
names = lnames.toArray(new String[lnames.size()]);
}
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/notifications/NewJobSender.java b/src/main/java/org/apache/sling/event/impl/jobs/notifications/NewJobSender.java
index 590026f..03c5769 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/notifications/NewJobSender.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/notifications/NewJobSender.java
@@ -22,10 +22,6 @@ import java.util.Dictionary;
import java.util.Hashtable;
import java.util.List;
-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.sling.api.resource.observation.ExternalResourceChangeListener;
import org.apache.sling.api.resource.observation.ResourceChange;
import org.apache.sling.api.resource.observation.ResourceChange.ChangeType;
@@ -36,6 +32,10 @@ import org.apache.sling.event.jobs.NotificationConstants;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.slf4j.Logger;
@@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory;
* This component receives resource added events and sends a job
* created event.
*/
-@Component
+@Component(service = {})
public class NewJobSender implements ResourceChangeListener, ExternalResourceChangeListener {
/** Logger. */
@@ -68,7 +68,7 @@ public class NewJobSender implements ResourceChangeListener, ExternalResourceCha
*/
@Activate
protected void activate(final BundleContext bundleContext) {
- final Dictionary<String, Object> properties = new Hashtable<String, Object>();
+ final Dictionary<String, Object> properties = new Hashtable<>();
properties.put(Constants.SERVICE_DESCRIPTION, "Apache Sling Job Topic Manager Event Handler");
properties.put(Constants.SERVICE_VENDOR, "The Apache Software Foundation");
properties.put(ResourceChangeListener.CHANGES, ChangeType.ADDED.toString());
@@ -104,7 +104,7 @@ public class NewJobSender implements ResourceChangeListener, ExternalResourceCha
if ( path.indexOf("_", topicEnd + 1) != -1 ) {
// only job id and topic are guaranteed
- final Dictionary<String, Object> properties = new Hashtable<String, Object>();
+ final Dictionary<String, Object> properties = new Hashtable<>();
properties.put(NotificationConstants.NOTIFICATION_PROPERTY_JOB_ID, jobId);
properties.put(NotificationConstants.NOTIFICATION_PROPERTY_JOB_TOPIC, topic);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java b/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
index fd7fcf8..7141984 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
@@ -29,13 +29,6 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
-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.Properties;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
import org.apache.sling.commons.scheduler.Scheduler;
@@ -48,7 +41,6 @@ import org.apache.sling.event.impl.jobs.JobImpl;
import org.apache.sling.event.impl.jobs.config.ConfigurationChangeListener;
import org.apache.sling.event.impl.jobs.config.InternalQueueConfiguration;
import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration;
-import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager;
import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager.QueueInfo;
import org.apache.sling.event.impl.jobs.jmx.QueueStatusEvent;
import org.apache.sling.event.impl.jobs.jmx.QueuesMBeanImpl;
@@ -59,6 +51,11 @@ import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.NotificationConstants;
import org.apache.sling.event.jobs.Queue;
import org.apache.sling.event.jobs.jmx.QueuesMBean;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.event.EventConstants;
@@ -70,13 +67,14 @@ import org.slf4j.LoggerFactory;
/**
* Implementation of the queue manager.
*/
-@Component(immediate=true)
-@Service(value={Runnable.class, QueueManager.class, EventHandler.class})
-@Properties({
- @Property(name=Scheduler.PROPERTY_SCHEDULER_PERIOD, longValue=60),
- @Property(name=Scheduler.PROPERTY_SCHEDULER_CONCURRENT, boolValue=false),
- @Property(name=EventConstants.EVENT_TOPIC, value=NotificationConstants.TOPIC_JOB_ADDED)
-})
+@Component(immediate=true,
+ service={Runnable.class, QueueManager.class, EventHandler.class},
+ property={
+ Scheduler.PROPERTY_SCHEDULER_PERIOD + ":Long=60",
+ Scheduler.PROPERTY_SCHEDULER_CONCURRENT + ":Boolean=false",
+ EventConstants.EVENT_TOPIC + "=" + NotificationConstants.TOPIC_JOB_ADDED,
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+ })
public class QueueManager
implements Runnable, EventHandler, ConfigurationChangeListener {
@@ -101,7 +99,7 @@ public class QueueManager
/**
* Our thread pool.
*/
- @Reference(referenceInterface=EventingThreadPool.class)
+ @Reference(service=EventingThreadPool.class)
private ThreadPool threadPool;
/** The job manager configuration. */
@@ -115,7 +113,7 @@ public class QueueManager
private final Object queuesLock = new Object();
/** All active queues. */
- private final Map<String, JobQueueImpl> queues = new ConcurrentHashMap<String, JobQueueImpl>();
+ private final Map<String, JobQueueImpl> queues = new ConcurrentHashMap<>();
/** We count the scheduler runs. */
private volatile long schedulerRuns;
@@ -294,7 +292,7 @@ public class QueueManager
private void restart() {
// let's rename/close all queues and clear them
synchronized ( queuesLock ) {
- final List<JobQueueImpl> queues = new ArrayList<JobQueueImpl>(this.queues.values());
+ final List<JobQueueImpl> queues = new ArrayList<>(this.queues.values());
for(final JobQueueImpl queue : queues ) {
this.outdateQueue(queue);
}
@@ -382,7 +380,7 @@ public class QueueManager
* Scan the resource tree for topics.
*/
private Set<String> scanTopics() {
- final Set<String> topics = new HashSet<String>();
+ final Set<String> topics = new HashSet<>();
final ResourceResolver resolver = this.configuration.createResourceResolver();
try {
@@ -420,12 +418,12 @@ public class QueueManager
* Get the latest mapping from queue name to topics
*/
private Map<QueueInfo, Set<String>> updateTopicMapping(final Set<String> topics) {
- final Map<QueueInfo, Set<String>> mapping = new HashMap<QueueConfigurationManager.QueueInfo, Set<String>>();
+ final Map<QueueInfo, Set<String>> mapping = new HashMap<>();
for(final String topic : topics) {
final QueueInfo queueInfo = this.configuration.getQueueConfigurationManager().getQueueInfo(topic);
Set<String> queueTopics = mapping.get(queueInfo);
if ( queueTopics == null ) {
- queueTopics = new HashSet<String>();
+ queueTopics = new HashSet<>();
mapping.put(queueInfo, queueTopics);
}
queueTopics.add(topic);
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/stats/StatisticsManager.java b/src/main/java/org/apache/sling/event/impl/jobs/stats/StatisticsManager.java
index c2d3d15..455cb1e 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/stats/StatisticsManager.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/stats/StatisticsManager.java
@@ -22,19 +22,21 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.event.impl.jobs.InternalJobState;
import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration;
import org.apache.sling.event.jobs.Statistics;
import org.apache.sling.event.jobs.TopicStatistics;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
/**
* The statistics manager keeps track of all statistics related tasks.
*/
-@Component
-@Service(value=StatisticsManager.class)
+@Component(service=StatisticsManager.class,
+ property = {
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
public class StatisticsManager {
/** The job manager configuration. */
@@ -56,10 +58,10 @@ public class StatisticsManager {
};
/** Statistics per topic. */
- private final ConcurrentMap<String, TopicStatistics> topicStatistics = new ConcurrentHashMap<String, TopicStatistics>();
+ private final ConcurrentMap<String, TopicStatistics> topicStatistics = new ConcurrentHashMap<>();
/** Statistics per queue. */
- private final ConcurrentMap<String, Statistics> queueStatistics = new ConcurrentHashMap<String, Statistics>();
+ private final ConcurrentMap<String, Statistics> queueStatistics = new ConcurrentHashMap<>();
/**
* Get the global statistics.
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask.java b/src/main/java/org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask.java
index 6cb136b..29feafa 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask.java
@@ -24,10 +24,6 @@ import java.util.Calendar;
import java.util.Iterator;
import java.util.List;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Property;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.Service;
import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
@@ -39,6 +35,9 @@ import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.consumer.JobExecutionContext;
import org.apache.sling.event.jobs.consumer.JobExecutionResult;
import org.apache.sling.event.jobs.consumer.JobExecutor;
+import org.osgi.framework.Constants;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,9 +51,11 @@ import org.slf4j.LoggerFactory;
* The value should either be a string or an array of string. Allowed values are:
* SUCCEEDED, STOPPED, GIVEN_UP, ERROR, DROPPED
*/
-@Component
-@Service(value = JobExecutor.class)
-@Property(name = JobExecutor.PROPERTY_TOPICS, value = "org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask")
+@Component(service = JobExecutor.class,
+ property = {
+ JobExecutor.PROPERTY_TOPICS + "=org/apache/sling/event/impl/jobs/tasks/HistoryCleanUpTask",
+ Constants.SERVICE_VENDOR + "=The Apache Software Foundation"
+})
public class HistoryCleanUpTask implements JobExecutor {
private static final String PROPERTY_AGE = "age";
@@ -88,7 +89,7 @@ public class HistoryCleanUpTask implements JobExecutor {
final List<String> stateList;
if ( states != null ) {
- stateList = new ArrayList<String>();
+ stateList = new ArrayList<>();
for(final String s : states) {
stateList.add(s);
}
diff --git a/src/test/java/org/apache/sling/event/impl/jobs/JobConsumerManagerTest.java b/src/test/java/org/apache/sling/event/impl/jobs/JobConsumerManagerTest.java
index f976767..a9d1b83 100644
--- a/src/test/java/org/apache/sling/event/impl/jobs/JobConsumerManagerTest.java
+++ b/src/test/java/org/apache/sling/event/impl/jobs/JobConsumerManagerTest.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
-import java.util.Collections;
+import java.lang.annotation.Annotation;
import org.apache.sling.event.jobs.consumer.JobConsumer;
import org.apache.sling.event.jobs.consumer.JobExecutor;
@@ -34,10 +34,34 @@ import org.osgi.framework.ServiceReference;
public class JobConsumerManagerTest {
+ private JobConsumerManager.Config getDefaultConfig() {
+ return new JobConsumerManager.Config() {
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return JobConsumerManager.Config.class;
+ }
+
+ @Override
+ public boolean org_apache_sling_installer_configuration_persist() {
+ return false;
+ }
+
+ @Override
+ public String[] job_consumermanager_whitelist() {
+ return new String[] {"*"};
+ }
+
+ @Override
+ public String[] job_consumermanager_blacklist() {
+ return null;
+ }
+ };
+ }
@Test public void testSimpleMappingConsumer() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobConsumer jc1 = Mockito.mock(JobConsumer.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -56,7 +80,7 @@ public class JobConsumerManagerTest {
@Test public void testCategoryMappingConsumer() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobConsumer jc1 = Mockito.mock(JobConsumer.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -75,7 +99,7 @@ public class JobConsumerManagerTest {
@Test public void testSubCategoryMappingConsumer() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobConsumer jc1 = Mockito.mock(JobConsumer.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -94,7 +118,7 @@ public class JobConsumerManagerTest {
@Test public void testSimpleMappingExecutor() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobExecutor jc1 = Mockito.mock(JobExecutor.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -113,7 +137,7 @@ public class JobConsumerManagerTest {
@Test public void testCategoryMappingExecutor() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobExecutor jc1 = Mockito.mock(JobExecutor.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -132,7 +156,7 @@ public class JobConsumerManagerTest {
@Test public void testSubCategoryMappingExecutor() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobExecutor jc1 = Mockito.mock(JobExecutor.class);
final ServiceReference ref1 = Mockito.mock(ServiceReference.class);
@@ -151,7 +175,7 @@ public class JobConsumerManagerTest {
@Test public void testRanking() {
final BundleContext bc = Mockito.mock(BundleContext.class);
final JobConsumerManager jcs = new JobConsumerManager();
- jcs.activate(bc, Collections.EMPTY_MAP);
+ jcs.activate(bc, getDefaultConfig());
final JobExecutor jc1 = Mockito.mock(JobExecutor.class);
final JobExecutor jc2 = Mockito.mock(JobExecutor.class);
diff --git a/src/test/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfigurationTest.java b/src/test/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfigurationTest.java
index d485b2b..817407f 100644
--- a/src/test/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfigurationTest.java
+++ b/src/test/java/org/apache/sling/event/impl/jobs/config/InternalQueueConfigurationTest.java
@@ -24,68 +24,128 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import java.lang.annotation.Annotation;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class InternalQueueConfigurationTest {
- @org.junit.Test public void testMaxParallel() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_NAME, "QueueConfigurationTest");
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, -1);
+ private InternalQueueConfiguration.Config createConfig(final double maxParallel) {
+ return createConfig(null, "QueueConfigurationTest", maxParallel);
+ }
+
+ private InternalQueueConfiguration.Config createConfig(final String[] topics) {
+ return createConfig(topics, "QueueConfigurationTest", ConfigurationConstants.DEFAULT_MAX_PARALLEL);
+ }
+
+ private InternalQueueConfiguration.Config createConfig(final String[] topics, final String name) {
+ return createConfig(topics, name, ConfigurationConstants.DEFAULT_MAX_PARALLEL);
+ }
+
+ private InternalQueueConfiguration.Config createConfig(final String[] topics,
+ final String name,
+ final double maxParallel) {
+ return new InternalQueueConfiguration.Config() {
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return InternalQueueConfiguration.Config.class;
+ }
+
+ @Override
+ public String queue_name() {
+ return name;
+ }
+
+ @Override
+ public String[] queue_topics() {
+ return topics;
+ }
+
+ @Override
+ public String queue_type() {
+ return "UNORDERED";
+ }
+
+ @Override
+ public String queue_priority() {
+ return ConfigurationConstants.DEFAULT_PRIORITY;
+ }
+
+ @Override
+ public int queue_retries() {
+ return ConfigurationConstants.DEFAULT_RETRIES;
+ }
+
+ @Override
+ public long queue_retrydelay() {
+ return ConfigurationConstants.DEFAULT_RETRY_DELAY;
+ }
+
+ @Override
+ public double queue_maxparallel() {
+ return maxParallel;
+ }
+
+ @Override
+ public boolean queue_keepJobs() {
+ return false;
+ }
+
+ @Override
+ public boolean queue_preferRunOnCreationInstance() {
+ return false;
+ }
+
+ @Override
+ public int queue_threadPoolSize() {
+ return 0;
+ }
+
+ @Override
+ public int service_ranking() {
+ return 0;
+ }
+ };
+ }
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ @org.junit.Test public void testMaxParallel() {
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(-1));
assertEquals(Runtime.getRuntime().availableProcessors(), c.getMaxParallel());
// Edge cases 0.0 and 1.0 (treated as int numbers)
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 0.0);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(0.0));
assertEquals(0, c.getMaxParallel());
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 1.0);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(1.0));
assertEquals(1, c.getMaxParallel());
// percentage (50%)
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 0.5);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(0.5));
assertEquals((int) Math.round(Runtime.getRuntime().availableProcessors() * 0.5), c.getMaxParallel());
// rounding
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 0.90);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(0.90));
assertEquals((int) Math.round(Runtime.getRuntime().availableProcessors() * 0.9), c.getMaxParallel());
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 0.99);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(0.99));
assertEquals((int) Math.round(Runtime.getRuntime().availableProcessors() * 0.99), c.getMaxParallel());
// Percentages can't go over 99% (0.99)
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, 1.01);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(1.01));
assertEquals(Runtime.getRuntime().availableProcessors(), c.getMaxParallel());
// Treat negative values same a -1 (all cores)
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, -0.5);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(-0.5));
assertEquals(Runtime.getRuntime().availableProcessors(), c.getMaxParallel());
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, -2);
- c = InternalQueueConfiguration.fromConfiguration(p);
+ c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(-2));
assertEquals(Runtime.getRuntime().availableProcessors(), c.getMaxParallel());
-
- // Invalid number results in ConfigurationConstants.DEFAULT_MAX_PARALLEL
- p.put(ConfigurationConstants.PROP_MAX_PARALLEL, "a string");
- c = InternalQueueConfiguration.fromConfiguration(p);
- assertEquals(ConfigurationConstants.DEFAULT_MAX_PARALLEL, c.getMaxParallel());
}
@org.junit.Test public void testTopicMatchersDot() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a."});
- p.put(ConfigurationConstants.PROP_NAME, "test");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a."}));
assertTrue(c.isValid());
assertNotNull(c.match("a/b"));
assertNotNull(c.match("a/c"));
@@ -96,11 +156,7 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatchersStar() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a*"});
- p.put(ConfigurationConstants.PROP_NAME, "test");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a*"}));
assertTrue(c.isValid());
assertNotNull(c.match("a/b"));
assertNotNull(c.match("a/c"));
@@ -111,11 +167,7 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatchers() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a"});
- p.put(ConfigurationConstants.PROP_NAME, "test");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a"}));
assertTrue(c.isValid());
assertNull(c.match("a/b"));
assertNull(c.match("a/c"));
@@ -126,11 +178,7 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatcherAndReplacement() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a."});
- p.put(ConfigurationConstants.PROP_NAME, "test-queue-{0}");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a."}, "test-queue-{0}"));
assertTrue(c.isValid());
final String b = "a/b";
assertNotNull(c.match(b));
@@ -141,11 +189,7 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatchersDotAndSlash() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a/."});
- p.put(ConfigurationConstants.PROP_NAME, "test");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a/."}));
assertTrue(c.isValid());
assertNotNull(c.match("a/b"));
assertNotNull(c.match("a/c"));
@@ -156,11 +200,11 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatchersStarAndSlash() {
- final Map<String, Object> p = new HashMap<String, Object>();
+ final Map<String, Object> p = new HashMap<>();
p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a/*"});
p.put(ConfigurationConstants.PROP_NAME, "test");
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a/*"}));
assertTrue(c.isValid());
assertNotNull(c.match("a/b"));
assertNotNull(c.match("a/c"));
@@ -171,11 +215,7 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testTopicMatcherAndReplacementAndSlash() {
- final Map<String, Object> p = new HashMap<String, Object>();
- p.put(ConfigurationConstants.PROP_TOPICS, new String[] {"a/."});
- p.put(ConfigurationConstants.PROP_NAME, "test-queue-{0}");
-
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(new String[] {"a/."}, "test-queue-{0}"));
assertTrue(c.isValid());
final String b = "a/b";
assertNotNull(c.match(b));
@@ -186,10 +226,10 @@ public class InternalQueueConfigurationTest {
}
@org.junit.Test public void testNoTopicMatchers() {
- final Map<String, Object> p = new HashMap<String, Object>();
+ final Map<String, Object> p = new HashMap<>();
p.put(ConfigurationConstants.PROP_NAME, "test");
- InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(p);
+ InternalQueueConfiguration c = InternalQueueConfiguration.fromConfiguration(Collections.<String, Object>emptyMap(), createConfig(null));
assertFalse(c.isValid());
}
}
diff --git a/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java b/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
index 3a77364..eb3de77 100644
--- a/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
+++ b/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
@@ -293,10 +293,10 @@ public abstract class AbstractJobHandlingTest {
public void setup() throws IOException {
// set load delay to 3 sec
final org.osgi.service.cm.Configuration c2 = this.configAdmin.getConfiguration("org.apache.sling.event.impl.jobs.jcr.PersistenceHandler", null);
- Dictionary<String, Object> p2 = new Hashtable<String, Object>();
+ Dictionary<String, Object> p2 = new Hashtable<>();
p2.put(JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY, 3L);
// and startup.delay to 1sec - otherwise default of 30sec breaks tests!
- p2.put(JobManagerConfiguration.PROPERTY_STARTUP_DELAY, 1L);
+ p2.put("startup.delay", 1L);
c2.update(p2);
// SLING-5560 : since the above (re)config is now applied, we're safe
@@ -390,7 +390,7 @@ public abstract class AbstractJobHandlingTest {
*/
protected ServiceRegistration<EventHandler> registerEventHandler(final String topic,
final EventHandler handler) {
- final Dictionary<String, Object> props = new Hashtable<String, Object>();
+ final Dictionary<String, Object> props = new Hashtable<>();
props.put(EventConstants.EVENT_TOPIC, topic);
final ServiceRegistration<EventHandler> reg = this.bc.registerService(EventHandler.class,
handler, props);
@@ -429,7 +429,7 @@ public abstract class AbstractJobHandlingTest {
protected ServiceRegistration<JobConsumer> registerJobConsumer(final String topic,
final JobConsumer handler) {
long cc = this.getConsumerChangeCount();
- final Dictionary<String, Object> props = new Hashtable<String, Object>();
+ final Dictionary<String, Object> props = new Hashtable<>();
props.put(JobConsumer.PROPERTY_TOPICS, topic);
final ServiceRegistration<JobConsumer> reg = this.bc.registerService(JobConsumer.class,
handler, props);
@@ -444,7 +444,7 @@ public abstract class AbstractJobHandlingTest {
protected ServiceRegistration<JobExecutor> registerJobExecutor(final String topic,
final JobExecutor handler) {
long cc = this.getConsumerChangeCount();
- final Dictionary<String, Object> props = new Hashtable<String, Object>();
+ final Dictionary<String, Object> props = new Hashtable<>();
props.put(JobConsumer.PROPERTY_TOPICS, topic);
final ServiceRegistration<JobExecutor> reg = this.bc.registerService(JobExecutor.class,
handler, props);
--
To stop receiving notification emails like this one, please contact
"commits@sling.apache.org" <co...@sling.apache.org>.