Skip to content

Commit

Permalink
SLING-11831 - Allow setting job properties for custom job state (#26)
Browse files Browse the repository at this point in the history
* SLING-11831 - Allow setting job properties for custom job state

* SLING-11831 - Allow setting job properties for custom job state

* SLING-11831 - Allow setting job properties for custom job state

* Add NotNull annotations
  • Loading branch information
andrewmkhoury committed May 3, 2023
1 parent 9237b74 commit 1cb0ae2
Show file tree
Hide file tree
Showing 5 changed files with 198 additions and 4 deletions.
2 changes: 1 addition & 1 deletion bnd.bnd
Expand Up @@ -19,7 +19,7 @@ Export-Package: \
org.apache.sling.event.api

-includeresource:\
org.apache.sling.event.api-1.0.0.jar=org.apache.sling.event.api-1.0.0.jar;lib:=true,\
org.apache.sling.event.api-1.0.2.jar=org.apache.sling.event.api-1.0.2.jar;lib:=true,\
@quartz-[0-9.]*.jar!/org/quartz/CronExpression.class,\
@quartz-[0-9.]*.jar!/org/quartz/ValueSet.class,\
@org.apache.sling.commons.osgi-[0-9.]*.jar!/org/apache/sling/commons/osgi/PropertiesUtil.class,\
Expand Down
7 changes: 6 additions & 1 deletion pom.xml
Expand Up @@ -119,6 +119,11 @@
<groupId>org.osgi</groupId>
<artifactId>org.osgi.service.component.annotations</artifactId>
</dependency>
<dependency>
<groupId>org.jetbrains</groupId>
<artifactId>annotations</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.discovery.api</artifactId>
Expand Down Expand Up @@ -269,7 +274,7 @@
<dependency>
<groupId>org.apache.sling</groupId>
<artifactId>org.apache.sling.event.api</artifactId>
<version>1.0.0</version>
<version>1.0.2</version>
</dependency>

<dependency>
Expand Down
Expand Up @@ -24,6 +24,7 @@
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.jetbrains.annotations.NotNull;

/**
* Implementation of the job execution context passed to
Expand Down Expand Up @@ -85,7 +86,23 @@ public void updateProgress(final long eta) {
}

@Override
public void log(final String message, Object... args) {
public void setProperty(@NotNull final String name, final Object value) {
if ( name == null ) {
throw new IllegalArgumentException("Name must not be null");
}
if ( value == null ) {
throw new IllegalArgumentException("Value must not be null");
}
if ( name.startsWith("slingevent:") || name.startsWith(":slingevent:")) {
throw new IllegalArgumentException("Property name must not start with slingevent: or :slingevent: " + name);
}

handler.getJob().setProperty(name, value);
handler.persistJobProperties(name);
}

@Override
public void log(@NotNull final String message, Object... args) {
final int logMaxCount = handler.getProgressLogMaxCount();
handler.persistJobProperties(handler.getJob().log(logMaxCount, message, args));
}
Expand Down
Expand Up @@ -150,6 +150,10 @@ private void historyCleanUpRemovedJobs(Calendar since) {
HistoryCleanUpTask.cleanup(
since,
resolver,
/**
* We use a dummy context here as we are running it as a
* scheduled task and not as a job.
*/
new JobExecutionContext() {
@Override
public void asyncProcessingFinished(JobExecutionResult result) {
Expand Down Expand Up @@ -181,6 +185,11 @@ public void log(String message, Object... args) {

}

@Override
public void setProperty(String name, Object value) {

}

@Override
public ResultBuilder result() {
return new ResultBuilderImpl();
Expand Down Expand Up @@ -436,7 +445,7 @@ private void cleanUpInstanceIdFolders(final TopologyCapabilities caps, final Str
for(final Resource r : toDelete) {
if ( caps.isActive() ) {
resolver.delete(r);
resolver.commit();
resolver.commit();
}
}
}
Expand Down
@@ -0,0 +1,163 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.sling.event.impl.jobs.queues;

import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.math.BigInteger;
import java.util.HashMap;

import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
import org.apache.sling.api.resource.ValueMap;
import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.commons.threads.ThreadPoolManager;
import org.apache.sling.event.impl.jobs.JobConsumerManager;
import org.apache.sling.event.impl.jobs.JobHandler;
import org.apache.sling.event.impl.jobs.JobImpl;
import org.apache.sling.event.impl.jobs.JobManagerImpl;
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.config.TopologyCapabilities;
import org.apache.sling.event.impl.jobs.stats.StatisticsManager;
import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.JobManager;
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.apache.sling.event.jobs.jmx.QueuesMBean;
import org.apache.sling.testing.mock.sling.junit.SlingContext;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.LoggerFactory;

import com.codahale.metrics.MetricRegistry;

public class JobExecutionContextImplTest {

@Rule
public SlingContext context = new SlingContext();

private JobManager jobManager;
private JobManagerConfiguration configuration;

@Before
public void setUp() {
configuration = createMockJobManagerConfiguration();

QueueConfigurationManager queueConfigMgr = mock(QueueConfigurationManager.class);
QueueInfo info = new QueueInfo();
info.queueConfiguration = new InternalQueueConfiguration();
when(queueConfigMgr.getQueueInfo(anyString())).thenReturn(info);
when(configuration.getQueueConfigurationManager()).thenReturn(queueConfigMgr);

TopologyCapabilities capabilities = mock(TopologyCapabilities.class);
JobConsumerManager jobConsumerManager = mock(JobConsumerManager.class);
QueueManager qManager = mock(QueueManager.class);
ThreadPoolManager threadPoolManager = mock(ThreadPoolManager.class);
MetricRegistry metric = mock(MetricRegistry.class);
StatisticsManager statisticsManager = mock(StatisticsManager.class);
QueuesMBean queuesMBean = mock(QueuesMBean.class);
Scheduler scheduler = mock(Scheduler.class);

context.registerService(JobManagerConfiguration.class, configuration);
context.registerService(TopologyCapabilities.class, capabilities);
context.registerService(QueueConfigurationManager.class, queueConfigMgr);
context.registerService(MetricRegistry.class, metric);
context.registerService(QueueManager.class, qManager);
context.registerService(JobConsumerManager.class, jobConsumerManager);
context.registerService(ThreadPoolManager.class, threadPoolManager);
context.registerService(StatisticsManager.class, statisticsManager);
context.registerService(QueuesMBean.class, queuesMBean);
context.registerService(Scheduler.class, scheduler);
context.registerService(JobExecutor.class, new TestJobExecutor(), new HashMap<String, Object>() {{
put(JobExecutor.PROPERTY_TOPICS, "test");
}});

jobManager = new JobManagerImpl();
context.registerInjectActivateService(jobManager, new HashMap<String, Object>());
}

@Test
public void testSetProperty() {
// Create a job - it will be written to the mock jcr
Job job = jobManager.addJob("test", null);

// Process the job
JobExecutor je = new TestJobExecutor();
je.process(job, new JobExecutionContextImpl(new JobHandler((JobImpl) job, je, configuration), null));

// Retrieve the custom property
assertEquals("testValue", job.getProperty("test", String.class));

final String testValue;
Iterable<Resource> resources = context.resourceResolver().getResource("/var/eventing/jobs/assigned").getChildren();
ValueMap props = resources.iterator().next().adaptTo(ValueMap.class);
testValue = props.get("test", String.class);
assertEquals("testValue", testValue);
}

public class TestJobExecutor implements JobExecutor {

@Override
public JobExecutionResult process(Job job, JobExecutionContext context) {
context.setProperty("test", "testValue");
return context.result().message("TEST").succeeded();
}
}

private JobManagerConfiguration createMockJobManagerConfiguration() {
JobManagerConfiguration jobManagerConfig = mock(JobManagerConfiguration.class);

String jobsPath = "/var/eventing";

when(jobManagerConfig.getUniqueId(anyString())).then(new Answer<String>() {
@Override
public String answer(InvocationOnMock invocation) throws Throwable {
byte [] digest = java.security.MessageDigest.getInstance("md5").digest(String.valueOf(Math.random()).getBytes("UTF-8"));
BigInteger bigInt = new BigInteger(1, digest);
String hashtext = bigInt.toString(16);
return hashtext + "_" + String.valueOf((int)(Math.random()* 1000000));
}
});
when(jobManagerConfig.getScheduledJobsPath(false)).thenReturn(jobsPath + "/scheduled-jobs");
when(jobManagerConfig.getUniquePath(eq(null), anyString(), anyString(), eq(null))).then(
new Answer<String>() {
@Override
public String answer(InvocationOnMock invocation) throws Throwable {
String jobNodePath = jobsPath + "/jobs/assigned/" + invocation.getArgument(2);
return jobNodePath;
}
}
);
when(jobManagerConfig.getAuditLogger()).thenReturn(LoggerFactory.getLogger("org.apache.sling.event.jobs.audit"));
ResourceResolver resolver = context.resourceResolver();
when(jobManagerConfig.createResourceResolver()).thenReturn(resolver);
return jobManagerConfig;
}
}

0 comments on commit 1cb0ae2

Please sign in to comment.