Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - issue BATCH-773: Refactor and extend ExportedJobLauncher to JobOperator

Browse Source 
Removed old ExportedJobLauncher and replace where necessary
This commit is contained in:
dsyer
2008-08-17 16:24:07 +00:00
parent 533b3788d2
commit f355d497c3
5 changed files with 30 additions and 620 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/ExportedJobLauncher.java
View File

@@ -1,85 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed 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.springframework.batch.core.launch.support;
import java.util.Properties;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
/**
* Interface to expose for remote management of jobs. Similar to
* {@link JobLauncher}, but replaces {@link JobExecution} and
* {@link JobParameters} with Strings in return types and method parameters, so
* it can be inspected by remote clients like the jconsole from the JRE without
* any links to Spring Batch.
*
* @author Dave Syer
*
*/
public interface ExportedJobLauncher {
/**
* Launch a job with the given name.
*
* @param name the name of the job to launch
* @return a representation of the {@link JobExecution} returned by a
* {@link JobLauncher}.
*/
String run(String name);
/**
* Launch a job with the given name and parameters.
*
* @param name the name of the job to launch
* @return a representation of the {@link JobExecution} returned by a
* {@link JobLauncher}.
*/
String run(String name, String params);
/**
* Stop all running jobs.
*/
void stop();
/**
* Stop running jobs with the supplied name.
*/
void stop(String name);
/**
* Clear volatile memory of any jobs that are not running (and are therefore
* completed or failed).
*/
void clear();
/**
* Enquire if any jobs launched here are still running.
*
* @return true if any jobs are running.
*/
boolean isRunning();
/**
* Query statistics of jobs.
*
* @return properties representing last known state of jobs with this name
* (including those that may have finished)
*/
public Properties getStatistics(String name);
}

234
spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleExportedJobLauncher.java
View File

@@ -1,234 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed 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.springframework.batch.core.launch.support;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.Map.Entry;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.configuration.JobLocator;
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
import org.springframework.batch.core.converter.JobParametersConverter;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class SimpleExportedJobLauncher implements ExportedJobLauncher, InitializingBean {
private static final String SEPARATOR = "|";
private JobLauncher launcher;
private JobLocator jobLocator;
private Map<String, JobExecution> registry = new HashMap<String, JobExecution>();
private JobParametersConverter jobParametersConverter = new DefaultJobParametersConverter();
/*
* (non-Javadoc)
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(launcher, "JobLauncher must be provided.");
Assert.notNull(jobLocator, "JobLocator must be provided.");
}
/**
* Public setter for the {@link JobLauncher}.
* @param launcher the launcher to set
*/
public void setLauncher(JobLauncher launcher) {
this.launcher = launcher;
}
/**
* Public setter for the JobLocator.
* @param jobLocator the jobLocator to set
*/
public void setJobLocator(JobLocator jobLocator) {
this.jobLocator = jobLocator;
}
/**
* Public setter for the JobParametersFactory.
* @param jobParametersConverter the jobParametersFactory to set
*/
public void setJobParametersFactory(JobParametersConverter jobParametersConverter) {
this.jobParametersConverter = jobParametersConverter;
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher
* #getStatistics()
*/
public Properties getStatistics(String name) {
Properties result = new Properties();
int i = 0;
Set<String> keys = getKeyForName(name);
for (String key : keys) {
JobExecution execution = (JobExecution) registry.get(key);
addStatistics(result, execution, name + "." + i + ".");
i++;
}
return result;
}
/**
* @param result
* @param execution
*/
private void addStatistics(Properties result, JobExecution execution, String prefix) {
int i = 0;
for (StepExecution stepExecution : execution.getStepExecutions()) {
ExecutionContext executionContext = stepExecution.getExecutionContext();
for (Entry<String, Object> entry : executionContext.entrySet()) {
result.setProperty(prefix + "step" + i + "." + entry.getKey(), "" + entry.getValue());
}
}
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher
* #isRunning()
*/
public boolean isRunning() {
for (String key : registry.keySet()) {
JobExecution execution = (JobExecution) registry.get(key);
if (execution.isRunning()) {
return true;
}
}
return false;
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher
* #run(java.lang.String)
*/
public String run(String name) {
return run(name, null);
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher
* #run(java.lang.String, java.lang.String)
*/
public String run(String name, String params) {
Job job;
try {
job = jobLocator.getJob(name);
}
catch (NoSuchJobException e) {
return e.getClass().getName() + ": " + e.getMessage();
}
JobParameters jobParameters = new JobParameters();
if (params != null) {
jobParameters = jobParametersConverter.getJobParameters(PropertiesConverter.stringToProperties(params));
}
JobExecution execution;
try {
execution = launcher.run(job, jobParameters);
}
catch (JobExecutionException e) {
return e.getClass().getName() + ": " + e.getMessage();
}
registry.put(name + SEPARATOR + params, execution);
return execution.toString();
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher
* #stop()
*/
public void stop() {
for (String key : registry.keySet()) {
JobExecution execution = (JobExecution) registry.get(key);
execution.stop();
}
}
/*
* (non-Javadoc)
* @see org.springframework.batch.core.launch.support.ExportedJobLauncher#stop(java.lang.String)
*/
public void stop(String name) {
Set<String> keys = getKeyForName(name);
for (String key : keys) {
JobExecution execution = (JobExecution) registry.get(key);
execution.stop();
}
}
/*
* (non-Javadoc)
* @see org.springframework.batch.core.launch.support.ExportedJobLauncher#clear()
*/
public void clear() {
for (String key : registry.keySet()) {
JobExecution execution = (JobExecution) registry.get(key);
if (!execution.isRunning()) {
registry.remove(key);
}
}
}
/**
* @param key
* @return
*/
private Set<String> getKeyForName(String name) {
Set<String> result = new HashSet<String>();
for (String key : registry.keySet()) {
if (key.startsWith(name + SEPARATOR)) {
result.add(key);
}
}
return result;
}
}

272
spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleExportedJobLauncherTests.java
View File

@@ -1,272 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed 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.springframework.batch.core.launch.support;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.configuration.support.MapJobRegistry;
import org.springframework.batch.core.configuration.support.ReferenceJobFactory;
import org.springframework.batch.core.converter.JobParametersConverter;
import org.springframework.batch.core.job.JobSupport;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.step.StepSupport;
import org.springframework.batch.item.ExecutionContext;
/**
* @author Dave Syer
*
*/
public class SimpleExportedJobLauncherTests {
private SimpleExportedJobLauncher launcher = new SimpleExportedJobLauncher();
private MapJobRegistry jobLocator = new MapJobRegistry();
private List<JobParameters> parameters = new ArrayList<JobParameters>();
private List<JobExecution> executions = new ArrayList<JobExecution>();
@Before
public void setUp() throws Exception {
launcher.setLauncher(new JobLauncher() {
public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException {
JobExecution result = new JobExecution(null);
StepExecution stepExecution = result.createStepExecution(new StepSupport("stepName"));
stepExecution.setExecutionContext(new ExecutionContext() {
{
put("foo", "bar");
}
});
parameters.add(jobParameters);
executions.add(result);
return result;
}
});
launcher.setJobLocator(jobLocator);
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#afterPropertiesSet()}
* .
*
* @throws Exception
*/
@Test
public void testAfterPropertiesSet() throws Exception {
launcher = new SimpleExportedJobLauncher();
try {
launcher.afterPropertiesSet();
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {
String message = e.getMessage();
assertTrue("Message does not contain 'launcher': " + message,
contains(message.toLowerCase(), "joblauncher"));
}
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#afterPropertiesSet()}
* .
*
* @throws Exception
*/
@Test
public void testAfterPropertiesSetWithLauncher() throws Exception {
launcher = new SimpleExportedJobLauncher();
launcher.setLauncher(new JobLauncher() {
public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException {
return null;
}
});
try {
launcher.afterPropertiesSet();
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {
String message = e.getMessage();
assertTrue("Message does not contain 'locator': " + message, contains(message.toLowerCase(), "joblocator"));
}
}
@Test
public void testGetStatistics() {
Properties props = launcher.getStatistics("foo");
assertNotNull(props);
assertEquals(0, props.entrySet().size());
}
@Test
public void testGetStatisticsWithContent() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.run("foo");
Properties props = launcher.getStatistics("foo");
assertNotNull(props);
assertEquals(1, props.entrySet().size());
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#isRunning()}
* .
*
* @throws Exception
*/
@Test
public void testIsRunning() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.run("foo");
assertTrue(launcher.isRunning());
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#isRunning()}
* .
*
* @throws Exception
*/
@Test
public void testAlreadyRunning() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.setLauncher(new JobLauncher() {
public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException {
throw new JobExecutionAlreadyRunningException("Bad!");
}
});
String value = launcher.run("foo");
assertTrue("Return value was not an exception: " + value,
contains(value, "JobExecutionAlreadyRunningException"));
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#run(java.lang.String)}
* .
*/
@Test
public void testRunNonExistentJob() {
String value = launcher.run("foo");
assertTrue("Return value was not an exception: " + value, contains(value, "NoSuchJobException"));
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#run(java.lang.String)}
* .
*
* @throws Exception
*/
@Test
public void testRunJobWithParameters() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
String value = launcher.run("foo", "bar=spam,bucket=crap");
assertTrue(launcher.isRunning());
assertTrue("Return value was not a JobExecution: " + value, contains(value, "JobExecution"));
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#run(java.lang.String)}
* .
*
* @throws Exception
*/
@Test
public void testRunJobWithParametersAndFactory() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.setJobParametersFactory(new JobParametersConverter() {
public JobParameters getJobParameters(Properties properties) {
return new JobParametersBuilder().addString("foo", "spam").toJobParameters();
}
public Properties getProperties(JobParameters params) {
return null;
}
});
launcher.run("foo", "bar=spam,bucket=crap");
assertEquals(1, parameters.size());
assertEquals("spam", parameters.get(0).getString("foo"));
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#stop()}
* .
*
* @throws Exception
*/
@Test
public void testStop() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.run("foo");
assertTrue(launcher.isRunning());
launcher.stop();
assertEquals(BatchStatus.STOPPING, executions.get(0).getStatus());
assertTrue(launcher.isRunning());
}
/**
* Test method for
* {@link org.springframework.batch.core.launch.support.SimpleExportedJobLauncher#stop()}
* .
*
* @throws Exception
*/
@Test
public void testStopWithName() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.run("foo");
assertTrue(launcher.isRunning());
launcher.stop("foo");
assertEquals(BatchStatus.STOPPING, executions.get(0).getStatus());
assertTrue(launcher.isRunning());
}
@Test
public void testClear() throws Exception {
jobLocator.register(new ReferenceJobFactory(new JobSupport("foo")));
launcher.run("foo");
// mark it as actually ended
executions.get(0).setEndTime(new Date());
launcher.clear();
assertFalse(launcher.isRunning());
}
private boolean contains(String str, String searchStr) {
return str.indexOf(searchStr) != -1;
}
}

28
spring-batch-samples/src/main/resources/adhoc-job-launcher-context.xml
View File

@@ -6,16 +6,10 @@
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-2.0.xsd">
<import resource="simple-job-launcher-context.xml" />
<bean class="org.springframework.jmx.export.MBeanExporter">
<property name="beans">
<map>
<entry key="spring:service=batch,bean=jobLauncher">
<bean class="org.springframework.batch.core.launch.support.SimpleExportedJobLauncher">
<property name="launcher" ref="jobLauncher" />
<property name="jobLocator" ref="jobRegistry" />
</bean>
</entry>
<entry key="spring:service=batch,bean=jobOperator" value-ref="jobOperator" />
<entry key="spring:service=batch,bean=notificationPublisher" value-ref="notificationPublisher" />
<entry key="spring:service=batch,bean=jobLoader" value-ref="loader" />
</map>
@@ -24,7 +18,7 @@
<bean class="org.springframework.jmx.export.assembler.InterfaceBasedMBeanInfoAssembler">
<property name="interfaceMappings">
<map>
<entry key="spring:service=batch,bean=jobLauncher" value="org.springframework.batch.core.launch.support.ExportedJobLauncher" />
<entry key="spring:service=batch,bean=jobLauncher" value="org.springframework.batch.core.launch.JobOperator" />
<entry key="spring:service=batch,bean=jobLoader" value="org.springframework.batch.sample.launch.JobLoader" />
</map>
</property>
@@ -33,10 +27,20 @@
</bean>
<bean id="notificationPublisher" class="org.springframework.batch.sample.jmx.JobExecutionNotificationPublisher" />
<bean id="jobRegistry" class="org.springframework.batch.core.configuration.support.MapJobRegistry" />
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
<property name="jobRepository" ref="jobRepository" />
<property name="taskExecutor">
<bean class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
<bean id="jobOperator" class="org.springframework.batch.core.launch.support.SimpleJobOperator">
<property name="jobExplorer">
<bean class="org.springframework.batch.core.explore.support.JobExplorerFactoryBean">
<property name="dataSource" ref="dataSource" />
<property name="databaseType" value="${environment}" />
</bean>
</property>
<property name="jobRegistry" ref="jobRegistry" />
<property name="jobLauncher">
<bean parent="jobLauncher">
<property name="taskExecutor">
<bean class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
</property>
</bean>
</property>
</bean>
<bean id="loader" class="org.springframework.batch.sample.launch.DefaultJobLoader">

31
spring-batch-samples/src/test/java/org/springframework/batch/sample/launch/RemoteLauncherTests.java
View File

@@ -16,8 +16,8 @@
package org.springframework.batch.sample.launch;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.List;
@@ -29,7 +29,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.launch.support.ExportedJobLauncher;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.launch.support.JobRegistryBackgroundJobRunner;
import org.springframework.jmx.MBeanServerNotFoundException;
import org.springframework.jmx.access.InvalidInvocationException;
@@ -46,7 +47,7 @@ public class RemoteLauncherTests {
private static List<Exception> errors = new ArrayList<Exception>();
private static ExportedJobLauncher launcher;
private static JobOperator launcher;
private static JobLoader loader;
@@ -61,21 +62,19 @@ public class RemoteLauncherTests {
public void testLaunchBadJob() throws Exception {
assertEquals(0, errors.size());
assertTrue(isConnected());
String result = launcher.run("foo");
assertTrue("Should contain 'NoSuchJobException': " + result, result.indexOf("NoSuchJobException") >= 0);
try {
launcher.start("foo", "");
fail("Expected NoSuchJobException");
} catch (NoSuchJobException e) {
//expected;
}
}
@Test
public void testLaunchAndStopRealJob() throws Exception {
public void testAvailableJobs() throws Exception {
assertEquals(0, errors.size());
assertTrue(isConnected());
String result = launcher.run("loopJob");
assertTrue("Should contain 'JobExecution': " + result, result.indexOf("JobExecution: id=") >= 0);
Thread.sleep(500);
assertTrue(launcher.isRunning());
launcher.stop();
Thread.sleep(500);
assertFalse(launcher.isRunning());
assertTrue(launcher.getJobNames().contains("loopJob"));
}
/*
@@ -112,9 +111,8 @@ public class RemoteLauncherTests {
}
if (launcher == null) {
MBeanServerConnectionFactoryBean connectionFactory = new MBeanServerConnectionFactoryBean();
connectionFactory.setServiceUrl("service:jmx:rmi://localhost/jndi/rmi://localhost:1099/batch-samples");
try {
launcher = (ExportedJobLauncher) getMBean(connectionFactory, "spring:service=batch,bean=jobLauncher", ExportedJobLauncher.class);
launcher = (JobOperator) getMBean(connectionFactory, "spring:service=batch,bean=jobOperator", JobOperator.class);
loader = (JobLoader) getMBean(connectionFactory, "spring:service=batch,bean=jobLoader", JobLoader.class);
}
catch (MBeanServerNotFoundException e) {
@@ -123,7 +121,7 @@ public class RemoteLauncherTests {
}
}
try {
launcher.isRunning();
launcher.getJobNames();
connected = loader.getConfigurations().size()>0;
logger.info("Configurations loaded: " + loader.getConfigurations());
}
@@ -139,7 +137,6 @@ public class RemoteLauncherTests {
factory.setObjectName(objectName);
factory.setProxyInterface(interfaceType);
factory.setServer((MBeanServerConnection) connectionFactory.getObject());
// factory.setServiceUrl("service:jmx:rmi://localhost/jndi/rmi://localhost:1099/batch-samples");
factory.afterPropertiesSet();
return factory.getObject();
}