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

IN PROGRESS - issue BATCH-159: JobExecutor should return a JobExecution (which itself contains the ExitStatus)

Browse Source 
http://opensource.atlassian.com/projects/spring/browse/BATCH-159

Merged JobLauncher implementations together and separated out thread interruption and JMX notification responsibilities to separate collaborators.
This commit is contained in:
dsyer
2007-10-24 17:35:05 +00:00
parent 533209a14b
commit 5b2b92da18
15 changed files with 1044 additions and 777 deletions
Download Patch File Download Diff File Expand all files Collapse all files

355
execution/src/main/java/org/springframework/batch/execution/bootstrap/AbstractJobLauncher.java
View File

@@ -1,355 +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.execution.bootstrap;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.runtime.JobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.execution.facade.NoSuchJobExecutionException;
import org.springframework.batch.execution.runtime.ScheduledJobIdentifierFactory;
import org.springframework.batch.io.exception.BatchConfigurationException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.Assert;
/**
* Base class for {@link JobLauncher} implementations making no choices about
* concurrent processing of jobs.
*
* @see JobLauncher
* @author Dave Syer
*/
public abstract class AbstractJobLauncher implements JobLauncher,
InitializingBean, ApplicationListener {
private static final Log logger = LogFactory
.getLog(AbstractJobLauncher.class);
protected JobExecutorFacade jobExecutorFacade;
private String jobConfigurationName;
private final Object monitor = new Object();
// Do not autostart by default - allow user to set job configuration
// later and then manually start:
private volatile boolean autoStart = false;
private JobIdentifierFactory jobIdentifierFactory = new ScheduledJobIdentifierFactory();
// A private registry for keeping track of running jobs.
private volatile Map registry = new HashMap();
/**
* Setter for {@link JobIdentifierFactory}.
*
* @param jobIdentifierFactory
* the {@link JobIdentifierFactory} to set
*/
public void setJobIdentifierFactory(
JobIdentifierFactory jobIdentifierFactory) {
this.jobIdentifierFactory = jobIdentifierFactory;
}
/**
* Setter for the {@link JobConfiguration} that this launcher will run.
*
* @param jobConfiguration
* the jobConfiguration to set
*/
public void setJobConfigurationName(String jobConfiguration) {
this.jobConfigurationName = jobConfiguration;
}
/**
* Setter for autostart flag. If this is true then the container will be
* started when the Spring context is refreshed. Defaults to false.
*
* @param autoStart
*/
public void setAutoStart(boolean autoStart) {
this.autoStart = autoStart;
}
/**
* Setter for {@link JobExecutorFacade}. Mandatory property.
*
* @param batchContainer
*/
public void setJobExecutorFacade(JobExecutorFacade jobExecutorFacade) {
this.jobExecutorFacade = jobExecutorFacade;
}
/**
* Check that mandatory properties are set.
*
* @see #setJobExecutorFacade(JobExecutorFacade)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(jobExecutorFacade);
}
/**
* If autostart flag is on, initialise on context start-up and call
* {@link #run()}.
*
* @throws BatchConfigurationException
* if the job tries to but cannot start because of a
* {@link NoSuchJobConfigurationException}.
*
* @see org.springframework.context.ApplicationListener#onApplicationEvent(org.springframework.context.ApplicationEvent)
*
*/
public void onApplicationEvent(ApplicationEvent event) {
if ((event instanceof ContextRefreshedEvent) && this.autoStart
&& !isRunning()) {
try {
run();
} catch (NoSuchJobConfigurationException e) {
throw new BatchConfigurationException(
"Cannot start job on context refresh", e);
}
}
}
/**
* Extension point for subclasses. Implementations might choose to start the
* job in a new thread or in the current thread.<br/>
*
* @param jobIdentifier
* the identifier of the job to run
* @param exitCallback
* a callback that should be called by the implementation after
* the job has ended (or failed)
*
* @return an {@link ExitStatus} indicating the current knowledge of the
* state of the job.
*
* @param runtimeInformation
* the {@link JobIdentifier} to start the launcher with.
* @throws NoSuchJobConfigurationException
*/
protected abstract ExitStatus doRun(JobIdentifier jobIdentifier,
Runnable exitCallback) throws NoSuchJobConfigurationException;
/**
* Start the provided {@link JobIdentifier}.
*
* @throws NoSuchJobConfigurationException
* if the container cannot locate a job configuration
* @throws IllegalStateException
* if JobConfiguration is null.
* @see Lifecycle#start().
*/
public ExitStatus run(final JobIdentifier jobIdentifier)
throws NoSuchJobConfigurationException {
synchronized (monitor) {
if (isInternalRunning(jobIdentifier)) {
return ExitStatus.RUNNING;
}
}
register(jobIdentifier);
return doRun(jobIdentifier, new Runnable() {
public void run() {
unregister(jobIdentifier);
}
});
/*
* Subclasses don't explicitly have to take care of unregistering the
* jobIdentifier - they just have to call the exitCallback when the job
* is finished.
*/
}
/**
* Start a job execution with the given name. If a job is already running
* has no effect.
*
* @param name
* the name to assign to the job
* @throws NoSuchJobConfigurationException
*/
public ExitStatus run(String name) throws NoSuchJobConfigurationException {
if (name == null) {
throw new NoSuchJobConfigurationException(
"Null job name cannot be located.");
}
JobIdentifier runtimeInformation = jobIdentifierFactory
.getJobIdentifier(name);
return this.run(runtimeInformation);
}
/**
* Start a job execution with default name and other runtime information
* provided by the factory. If a job is already running has no effect. The
* default name is taken from the enclosed {@link JobConfiguration}.
*
* @throws NoSuchJobConfigurationException
*
* @throws NoSuchJobConfigurationException
* if the job configuration cannot be located
*
* @see #setJobIdentifierFactory(JobIdentifierFactory)
* @see org.springframework.context.Lifecycle#start()
*/
public ExitStatus run() throws NoSuchJobConfigurationException {
if (jobConfigurationName != null) {
return this.run(jobConfigurationName);
}
throw new NoSuchJobConfigurationException(
"Null default job name cannot be located.");
}
/**
* Extension point for subclasses to stop a specific job.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.container.bootstrap.BatchContainerLauncher#stop(JobRuntimeInformation))
*/
protected abstract void doStop(JobIdentifier runtimeInformation)
throws NoSuchJobExecutionException;
/**
* Stop all jobs if any are running. If not, no action will be taken.
* Delegates to the {@link #doStop()} method.
*
* @throws NoSuchJobExecutionException
* @see org.springframework.context.Lifecycle#stop()
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop()
*/
final public void stop() {
for (Iterator iter = new HashSet(registry.keySet()).iterator(); iter
.hasNext();) {
JobIdentifier context = (JobIdentifier) iter.next();
try {
stop(context);
} catch (NoSuchJobExecutionException e) {
logger.error(e);
}
}
}
/**
* Stop a job with this {@link JobIdentifier}. Delegates to the
* {@link #doStop(JobIdentifier)} method.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop(org.springframework.batch.core.domain.JobIdentifier)
* @see BatchContainer#stop(JobRuntimeInformation))
*/
final public void stop(JobIdentifier runtimeInformation)
throws NoSuchJobExecutionException {
synchronized (monitor) {
doStop(runtimeInformation);
}
}
/**
* Stop all jobs with {@link JobIdentifier} having this name. Delegates to
* the {@link #stop(JobIdentifier)}.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop(java.lang.String)
*/
final public void stop(String name) throws NoSuchJobExecutionException {
this.stop(jobIdentifierFactory.getJobIdentifier(name));
}
/**
* Check each registered {@link JobIdentifier} to see if it is running (@see
* {@link #isRunning(JobIdentifier)}), and if any are, then return true.
*
* @see org.springframework.batch.container.bootstrap.BatchContainerLauncher#isRunning()
*/
final public boolean isRunning() {
Collection jobs = new HashSet(registry.keySet());
for (Iterator iter = jobs.iterator(); iter.hasNext();) {
JobIdentifier jobIdentifier = (JobIdentifier) iter.next();
if (isInternalRunning(jobIdentifier)) {
return true;
}
}
return !jobs.isEmpty();
}
private boolean isInternalRunning(JobIdentifier jobIdentifier) {
synchronized (registry) {
return isRunning(jobIdentifier)
&& registry.containsKey(jobIdentifier);
}
}
/**
* Extension point for subclasses to check an individual
* {@link JobIdentifier} to see if it is running. As long as at least one
* job is running the launcher is deemed to be running.
*
* @param jobIdentifier
* a {@link JobIdentifier}
* @return always true. Subclasses can override and provide more accurate
* information.
*/
protected boolean isRunning(JobIdentifier jobIdentifier) {
return true;
}
/**
* Convenient synchronized accessor for the registry. Can be used by
* subclasses if necessary (but it isn't likely).
*
* @param jobIdentifier
*/
private void register(JobIdentifier jobIdentifier) {
synchronized (registry) {
registry.put(jobIdentifier, jobIdentifier);
}
}
/**
* Convenient synchronized accessor for the registry. Must be used by
* subclasses to release the {@link JobIdentifier} when a job is finished
* (or stopped).
*
* @param jobIdentifier
*/
private void unregister(JobIdentifier jobIdentifier) {
synchronized (registry) {
registry.remove(jobIdentifier);
}
}
}

96
execution/src/main/java/org/springframework/batch/execution/bootstrap/JobExecutionNotificationPublisher.java Normal file
View File

@@ -0,0 +1,96 @@
/*
* 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.execution.bootstrap;
import javax.management.Notification;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;
/**
* JMX notification broadcaster
*
* @author Dave Syer
* @since 2.1
*/
public class JobExecutionNotificationPublisher implements
ApplicationListener, NotificationPublisherAware {
private NotificationPublisher notificationPublisher;
private int notificationCount = 0;
/**
* Injection setter.
*
* @see org.springframework.jmx.export.notification.NotificationPublisherAware#setNotificationPublisher(org.springframework.jmx.export.notification.NotificationPublisher)
*/
public void setNotificationPublisher(
NotificationPublisher notificationPublisher) {
this.notificationPublisher = notificationPublisher;
}
/**
* If the event is a {@link RepeatOperationsApplicationEvent} for open and
* close we log the event at INFO level and send a JMX notification if we
* are also an MBean.
*
* @see org.springframework.batch.execution.bootstrap.SimpleJobLauncher#onApplicationEvent(org.springframework.context.ApplicationEvent)
*/
public void onApplicationEvent(ApplicationEvent applicationEvent) {
if (applicationEvent instanceof RepeatOperationsApplicationEvent) {
RepeatOperationsApplicationEvent event = (RepeatOperationsApplicationEvent) applicationEvent;
int type = event.getType();
if (type == RepeatOperationsApplicationEvent.OPEN
|| type == RepeatOperationsApplicationEvent.CLOSE
|| type == RepeatOperationsApplicationEvent.ERROR) {
String message = event.getMessage() + "; source="
+ event.getSource();
SimpleJobLauncher.logger.info(message);
publish(message);
}
return;
}
}
/**
* Publish the provided message to an external listener if there is one.
*
* @param message
* the message to publish
*/
private void publish(String message) {
if (notificationPublisher != null) {
Notification notification = new Notification(
"RepeatOperationsApplicationEvent", this,
notificationCount++, message);
/*
* We can't create a notification with a null source, but we can set
* it to null after creation(!). We want it to be null so that
* Spring will replace it automatically with the ObjectName (in
* ModelMBeanNotificationPublisher).
*/
notification.setSource(null);
notificationPublisher.sendNotification(notification);
}
}
}

515
execution/src/main/java/org/springframework/batch/execution/bootstrap/SimpleJobLauncher.java
View File

@@ -1,101 +1,414 @@
/*
* 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.execution.bootstrap;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.execution.facade.NoSuchJobExecutionException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.util.Assert;
/**
* Simple bootstrapping mechanism for running a single job execution in a
* {@link JobExecutorFacade}.
*
* <p>
* This simple implementation does not run the job asynchronously, so the start
* method will not return before the job ends. However, the job execution can be
* interrupted via the stop method in another thread.
* </p>
*
* @author Lucas Ward
* @author Dave Syer
* @since 2.1
*/
public class SimpleJobLauncher extends AbstractJobLauncher {
private volatile Thread processingThread;
private int running = 0;
/**
* Check whether or not the container is currently running. This is done by
* checking the thread to see if it is still alive.
*/
protected boolean isRunning(JobIdentifier jobIdentifier) {
return processingThread != null && processingThread.isAlive();
}
/**
* Start the provided facade. The current thread will first be saved. This
* may seem odd at first, however, this simple bootstrap requires that only
* one thread can kick off a container, and that the first thread that calls
* start is the 'processing thread'. If the container has already been
* started, no exception will be thrown.
*
* @throws NoSuchJobConfigurationException
* @see Lifecycle#start().
*
* @throws IllegalStateException
* if JobConfiguration is null.
*/
protected ExitStatus doRun(JobIdentifier jobIdentifier,
Runnable exitCallback) throws NoSuchJobConfigurationException {
Assert.notNull(jobIdentifier, "JobIdentifier must not be null.");
Assert.isTrue(running == 0,
"This launcher can run only one job at at time.");
/*
* There is no reason to kick off a new thread, since only one thread
* should be processing at once. However, a handle to the thread is
* maintained to allow for interrupt
*/
processingThread = Thread.currentThread();
try {
running++;
return jobExecutorFacade.start(jobIdentifier);
} finally {
running--;
exitCallback.run();
}
}
/**
* Interrupt the thread that is running the job.
*
* @see org.springframework.batch.execution.bootstrap.AbstractJobLauncher#doStop(org.springframework.batch.core.domain.JobIdentifier)
*/
protected void doStop(JobIdentifier runtimeInformation)
throws NoSuchJobExecutionException {
if (isRunning()) {
processingThread.interrupt();
}
}
}
/*
* 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.execution.bootstrap;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.runtime.JobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.execution.facade.NoSuchJobExecutionException;
import org.springframework.batch.execution.runtime.ScheduledJobIdentifierFactory;
import org.springframework.batch.io.exception.BatchConfigurationException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
/**
* Base class for {@link JobLauncher} implementations making no choices about
* concurrent processing of jobs.
*
* @see JobLauncher
* @author Dave Syer
*/
public class SimpleJobLauncher implements JobLauncher, InitializingBean,
ApplicationListener, ApplicationEventPublisherAware, StatisticsProvider {
protected static final Log logger = LogFactory
.getLog(SimpleJobLauncher.class);
protected JobExecutorFacade jobExecutorFacade;
private String jobConfigurationName;
private final Object monitor = new Object();
// Do not autostart by default - allow user to set job configuration
// later and then manually start:
private volatile boolean autoStart = false;
private JobIdentifierFactory jobIdentifierFactory = new ScheduledJobIdentifierFactory();
// A private registry for keeping track of running jobs.
private volatile Map registry = new HashMap();
private TaskExecutor taskExecutor = new SyncTaskExecutor();
ApplicationEventPublisher applicationEventPublisher;
/**
* Setter for {@link JobIdentifierFactory}.
*
* @param jobIdentifierFactory
* the {@link JobIdentifierFactory} to set
*/
public void setJobIdentifierFactory(
JobIdentifierFactory jobIdentifierFactory) {
this.jobIdentifierFactory = jobIdentifierFactory;
}
/**
* Setter for the {@link JobConfiguration} that this launcher will run.
*
* @param jobConfiguration
* the jobConfiguration to set
*/
public void setJobConfigurationName(String jobConfiguration) {
this.jobConfigurationName = jobConfiguration;
}
/**
* Setter for autostart flag. If this is true then the container will be
* started when the Spring context is refreshed. Defaults to false.
*
* @param autoStart
*/
public void setAutoStart(boolean autoStart) {
this.autoStart = autoStart;
}
/**
* Setter for {@link JobExecutorFacade}. Mandatory property.
*
* @param batchContainer
*/
public void setJobExecutorFacade(JobExecutorFacade jobExecutorFacade) {
this.jobExecutorFacade = jobExecutorFacade;
}
/**
* Check that mandatory properties are set.
*
* @see #setJobExecutorFacade(JobExecutorFacade)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(jobExecutorFacade);
}
/**
* If autostart flag is on, initialise on context start-up and call
* {@link #run()}.
*
* @throws BatchConfigurationException
* if the job tries to but cannot start because of a
* {@link NoSuchJobConfigurationException}.
*
* @see org.springframework.context.ApplicationListener#onApplicationEvent(org.springframework.context.ApplicationEvent)
*
*/
public void onApplicationEvent(ApplicationEvent event) {
if ((event instanceof ContextRefreshedEvent) && this.autoStart
&& !isRunning()) {
try {
run();
} catch (NoSuchJobConfigurationException e) {
throw new BatchConfigurationException(
"Cannot start job on context refresh", e);
}
}
}
/**
* Subclasses which need to make the call asynchronously should delegate to
* this method inside a Runnable or Callable, so that the internal
* housekeeping is done consistently.
*
* @param jobIdentifier
* @return
* @throws NoSuchJobConfigurationException
*/
protected final ExitStatus runInternal(final JobIdentifier jobIdentifier)
throws NoSuchJobConfigurationException {
synchronized (monitor) {
if (isInternalRunning(jobIdentifier)) {
return ExitStatus.RUNNING;
}
}
register(jobIdentifier);
try {
return jobExecutorFacade.start(jobIdentifier);
} finally {
unregister(jobIdentifier);
}
/*
* Subclasses don't explicitly have to take care of unregistering the
* jobIdentifier - they just have to call this method to make sure that
* it is done.
*/
}
/**
* Start a job execution with the given name. If a job is already running
* has no effect.
*
* @param name
* the name to assign to the job
* @throws NoSuchJobConfigurationException
*/
public ExitStatus run(String name) throws NoSuchJobConfigurationException {
if (name == null) {
throw new NoSuchJobConfigurationException(
"Null job name cannot be located.");
}
JobIdentifier runtimeInformation = jobIdentifierFactory
.getJobIdentifier(name);
return this.run(runtimeInformation);
}
/**
* Start a job execution with default name and other runtime information
* provided by the factory. If a job is already running has no effect. The
* default name is taken from the enclosed {@link JobConfiguration}.
*
* @throws NoSuchJobConfigurationException
*
* @throws NoSuchJobConfigurationException
* if the job configuration cannot be located
*
* @see #setJobIdentifierFactory(JobIdentifierFactory)
* @see org.springframework.context.Lifecycle#start()
*/
public ExitStatus run() throws NoSuchJobConfigurationException {
if (jobConfigurationName != null) {
return this.run(jobConfigurationName);
}
throw new NoSuchJobConfigurationException(
"Null default job name cannot be located.");
}
/**
* Extension point for subclasses to stop a specific job.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.container.bootstrap.BatchContainerLauncher#stop(JobRuntimeInformation))
*/
protected void doStop(JobIdentifier jobIdentifier)
throws NoSuchJobExecutionException {
jobExecutorFacade.stop(jobIdentifier);
}
/**
* Stop all jobs if any are running. If not, no action will be taken.
* Delegates to the {@link #doStop()} method.
*
* @throws NoSuchJobExecutionException
* @see org.springframework.context.Lifecycle#stop()
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop()
*/
final public void stop() {
for (Iterator iter = new HashSet(registry.keySet()).iterator(); iter
.hasNext();) {
JobIdentifier context = (JobIdentifier) iter.next();
try {
stop(context);
} catch (NoSuchJobExecutionException e) {
logger.error(e);
}
}
}
/**
* Stop a job with this {@link JobIdentifier}. Delegates to the
* {@link #doStop(JobIdentifier)} method.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop(org.springframework.batch.core.domain.JobIdentifier)
* @see BatchContainer#stop(JobRuntimeInformation))
*/
final public void stop(JobIdentifier runtimeInformation)
throws NoSuchJobExecutionException {
synchronized (monitor) {
doStop(runtimeInformation);
}
}
/**
* Stop all jobs with {@link JobIdentifier} having this name. Delegates to
* the {@link #stop(JobIdentifier)}.
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.batch.execution.bootstrap.JobLauncher#stop(java.lang.String)
*/
final public void stop(String name) throws NoSuchJobExecutionException {
this.stop(jobIdentifierFactory.getJobIdentifier(name));
}
/**
* Check each registered {@link JobIdentifier} to see if it is running (@see
* {@link #isRunning(JobIdentifier)}), and if any are, then return true.
*
* @see org.springframework.batch.container.bootstrap.BatchContainerLauncher#isRunning()
*/
final public boolean isRunning() {
Collection jobs = new HashSet(registry.keySet());
for (Iterator iter = jobs.iterator(); iter.hasNext();) {
JobIdentifier jobIdentifier = (JobIdentifier) iter.next();
if (isInternalRunning(jobIdentifier)) {
return true;
}
}
return !jobs.isEmpty();
}
private boolean isInternalRunning(JobIdentifier jobIdentifier) {
synchronized (registry) {
return isRunning(jobIdentifier)
&& registry.containsKey(jobIdentifier);
}
}
/**
* Extension point for subclasses to check an individual
* {@link JobIdentifier} to see if it is running. As long as at least one
* job is running the launcher is deemed to be running.
*
* @param jobIdentifier
* a {@link JobIdentifier}
* @return always true. Subclasses can override and provide more accurate
* information.
*/
protected boolean isRunning(JobIdentifier jobIdentifier) {
return true;
}
/**
* Convenient synchronized accessor for the registry. Can be used by
* subclasses if necessary (but it isn't likely).
*
* @param jobIdentifier
*/
private void register(JobIdentifier jobIdentifier) {
synchronized (registry) {
registry.put(jobIdentifier, jobIdentifier);
}
}
/**
* Convenient synchronized accessor for the registry. Must be used by
* subclasses to release the {@link JobIdentifier} when a job is finished
* (or stopped).
*
* @param jobIdentifier
*/
private void unregister(JobIdentifier jobIdentifier) {
synchronized (registry) {
registry.remove(jobIdentifier);
}
}
/**
* Setter for the {@link TaskExecutor}. Defaults to a
* {@link SyncTaskExecutor}.
*
* @param taskExecutor
* the taskExecutor to set
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Start the job using the task executor provided.
*
* @see org.springframework.batch.execution.bootstrap.SimpleJobLauncher#run(org.springframework.batch.core.domain.JobIdentifier)
*/
public ExitStatus run(final JobIdentifier jobIdentifier) {
Assert.state(taskExecutor != null, "TaskExecutor must be provided");
taskExecutor.execute(new Runnable() {
public void run() {
try {
runInternal(jobIdentifier);
} catch (NoSuchJobConfigurationException e) {
applicationEventPublisher
.publishEvent(new RepeatOperationsApplicationEvent(
jobIdentifier, "No such job",
RepeatOperationsApplicationEvent.ERROR));
logger.error(
"JobConfiguration could not be located inside Runnable for identifier: ["
+ jobIdentifier + "]", e);
}
}
});
return ExitStatus.UNKNOWN;
}
/**
* Accessor for the job executions passed back in response to a call to
* {@link #requestContextNotification()}. Because the request is
* potentially fulfilled asynchronously, and only on demand, the data might
* be out of date by the time this method is called, so it should be used
* for information purposes only.
*
* @return Properties representing the {@link JobExecution} objects passed
* up from the underlying execution. If there are no jobs running it
* will be empty.
*/
public Properties getStatistics() {
if (jobExecutorFacade instanceof StatisticsProvider) {
return ((StatisticsProvider) jobExecutorFacade).getStatistics();
} else {
return new Properties();
}
}
public void setApplicationEventPublisher(
ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
}

219
execution/src/main/java/org/springframework/batch/execution/bootstrap/TaskExecutorJobLauncher.java
View File

@@ -1,219 +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.execution.bootstrap;
import java.util.Properties;
import javax.management.Notification;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.execution.facade.NoSuchJobExecutionException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.ApplicationListener;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;
import org.springframework.util.Assert;
/**
* Bootstrapping mechanism for running job executions concurrently with a
* {@link JobExecutorFacade}.
*
* <p>
* This implementation can run jobs asynchronously. Jobs are stopped by calling
* the stop method in the {@link JobExecutorFacade}, which is a graceful
* shutdown.
* </p>
*
* @see JobExecutorFacade
* @author Dave Syer
* @since 2.1
*/
public class TaskExecutorJobLauncher extends AbstractJobLauncher implements
ApplicationListener, NotificationPublisherAware,
ApplicationEventPublisherAware {
private static final Log logger = LogFactory
.getLog(TaskExecutorJobLauncher.class);
private TaskExecutor taskExecutor = new SyncTaskExecutor();
private NotificationPublisher notificationPublisher;
private int notificationCount = 0;
private ApplicationEventPublisher applicationEventPublisher;
/*
* (non-Javadoc)
*
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
*/
public void setApplicationEventPublisher(
ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
/**
* Setter for the {@link TaskExecutor}. Defaults to a
* {@link SyncTaskExecutor}.
*
* @param taskExecutor
* the taskExecutor to set
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Injection setter.
*
* @see org.springframework.jmx.export.notification.NotificationPublisherAware#setNotificationPublisher(org.springframework.jmx.export.notification.NotificationPublisher)
*/
public void setNotificationPublisher(
NotificationPublisher notificationPublisher) {
this.notificationPublisher = notificationPublisher;
}
/**
* Start the job using the task executor provided. An {@link Runnable} is
* passed in by the caller which we need to call in a finally block.
*
* @throws NoSuchJobConfigurationException
* if a job configuration cannot be located.
*
* @see org.springframework.batch.execution.bootstrap.AbstractJobLauncher#doRun(org.springframework.batch.core.domain.JobIdentifier,
* java.lang.Runnable)
*/
protected ExitStatus doRun(final JobIdentifier jobIdentifier,
final Runnable exitCallback) {
Assert.state(taskExecutor != null, "TaskExecutor must be provided");
taskExecutor.execute(new Runnable() {
public void run() {
try {
jobExecutorFacade.start(jobIdentifier);
} catch (NoSuchJobConfigurationException e) {
applicationEventPublisher
.publishEvent(new RepeatOperationsApplicationEvent(
jobIdentifier, "No such job",
RepeatOperationsApplicationEvent.ERROR));
logger.error(
"JobConfiguration could not be located inside Runnable for identifier: ["
+ jobIdentifier + "]", e);
} finally {
exitCallback.run();
}
}
});
return ExitStatus.UNKNOWN;
}
/**
* Delegates to the underlying {@link JobExecutorFacade}. Does not wait for
* the jobs to stop (therefore returns immediately by default).
*
* @throws NoSuchJobExecutionException
*
* @see org.springframework.context.Lifecycle#stop()
*/
protected void doStop(JobIdentifier jobIdentifier)
throws NoSuchJobExecutionException {
jobExecutorFacade.stop(jobIdentifier);
// TODO: wait for the jobs to stop?
}
/**
* If the event is a {@link RepeatOperationsApplicationEvent} for open and
* close we log the event at INFO level and send a JMX notification if we
* are also an MBean.
*
* @see org.springframework.batch.execution.bootstrap.AbstractJobLauncher#onApplicationEvent(org.springframework.context.ApplicationEvent)
*/
public void onApplicationEvent(ApplicationEvent applicationEvent) {
super.onApplicationEvent(applicationEvent);
if (applicationEvent instanceof RepeatOperationsApplicationEvent) {
RepeatOperationsApplicationEvent event = (RepeatOperationsApplicationEvent) applicationEvent;
int type = event.getType();
if (type == RepeatOperationsApplicationEvent.OPEN
|| type == RepeatOperationsApplicationEvent.CLOSE
|| type == RepeatOperationsApplicationEvent.ERROR) {
String message = event.getMessage() + "; source="
+ event.getSource();
logger.info(message);
publish(message);
}
return;
}
}
/**
* Accessor for the job executions passed back in response to a call to
* {@link #requestContextNotification()}. Because the request is
* potentially fulfilled asynchronously, and only on demand, the data might
* be out of date by the time this method is called, so it should be used
* for information purposes only.
*
* @return Properties representing the {@link JobExecution} objects passed
* up from the underlying execution. If there are no jobs running it
* will be empty.
*/
public Properties getStatistics() {
if (jobExecutorFacade instanceof StatisticsProvider) {
return ((StatisticsProvider) jobExecutorFacade).getStatistics();
} else {
return new Properties();
}
}
/**
* Publish the provided message to an external listener if there is one.
*
* @param message
* the message to publish
*/
private void publish(String message) {
if (notificationPublisher != null) {
Notification notification = new Notification(
"RepeatOperationsApplicationEvent", this,
notificationCount++, message);
/*
* We can't create a notification with a null source, but we can set
* it to null after creation(!). We want it to be null so that
* Spring will replace it automatically with the ObjectName (in
* ModelMBeanNotificationPublisher).
*/
notification.setSource(null);
notificationPublisher.sendNotification(notification);
}
}
}

80
execution/src/main/java/org/springframework/batch/execution/bootstrap/support/ThreadInterruptJobExecutionListener.java Normal file
View File

@@ -0,0 +1,80 @@
/*
* 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.execution.bootstrap.support;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutionListenerSupport;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.util.Assert;
/**
* {@link JobExecutionListener} that will interrupt the Thread that the job was
* started in when the stop signal comes. Use only for a standalone process, not
* in an application server container.
*
* @author Dave Syer
*
*/
public class ThreadInterruptJobExecutionListener extends
JobExecutionListenerSupport {
private volatile Thread processingThread;
private int running = 0;
/**
* Save the current thread so it can be interrupted later. This may seem odd
* at first, however, a simple bootstrap requires that only one thread can
* kick off a container, and that the first thread that calls start is the
* 'processing thread'. If the container has already been started, no
* exception will be thrown.
*
* @see org.springframework.batch.execution.facade.JobExecutionListenerSupport#before(org.springframework.batch.core.domain.JobExecution)
*/
public void before(JobExecution execution) {
Assert.isTrue(running == 0,
"This listener only supports one job at at time.");
running++;
/*
* There is no reason to kick off a new thread, since only one thread
* should be processing at once. However, a handle to the thread is
* maintained to allow for interrupt
*/
processingThread = Thread.currentThread();
}
/**
* Interrupt the thread that is running the job if the {@link ExitStatus}
* indicates that it is still running.
*
* @see org.springframework.batch.execution.facade.JobExecutionListenerSupport#stop(org.springframework.batch.core.domain.JobExecution)
*/
public void stop(JobExecution execution) {
if (execution==null || execution.getExitStatus().isRunning()) {
processingThread.interrupt();
}
}
/**
* internal housekeeping.
*
* @see org.springframework.batch.execution.facade.JobExecutionListenerSupport#after(org.springframework.batch.core.domain.JobExecution)
*/
public void after(JobExecution execution) {
running--;
}
}

50
execution/src/main/java/org/springframework/batch/execution/facade/JobExecutionListener.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.execution.facade;
import org.springframework.batch.core.domain.JobExecution;
/**
* Listener interface for the job execution lifecycle.
*
* @author Dave Syer
*
*/
public interface JobExecutionListener {
/**
* Callback for the start of a job, before any steps are processed.
*
* @param execution
* the current {@link JobExecution}
*/
void before(JobExecution execution);
/**
* Callback for the start of a job, after all steps are processed, or on an
* error.
*
* @param execution
*/
void after(JobExecution execution);
/**
* Callback for a job that has been stopped, or asked to stop.
*
* @param execution
*/
void stop(JobExecution execution);
}

58
execution/src/main/java/org/springframework/batch/execution/facade/JobExecutionListenerSupport.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* 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.execution.facade;
import org.springframework.batch.core.domain.JobExecution;
/**
* Simple no-op implementation of {@link JobExecutionListener} which does
* nothing.
*
* @author Dave Syer
*
*/
public class JobExecutionListenerSupport implements JobExecutionListener {
/**
* No-op for subclasses to extend.
*
* @see org.springframework.batch.execution.facade.JobExecutionListener#after(org.springframework.batch.core.domain.JobExecution)
*/
public void after(JobExecution execution) {
// no-op
}
/**
* No-op for subclasses to extend.
*
* @see org.springframework.batch.execution.facade.JobExecutionListener#before(org.springframework.batch.core.domain.JobExecution)
*/
public void before(JobExecution execution) {
// no-op
}
/**
* No-op for subclasses to extend.
*
* @see org.springframework.batch.execution.facade.JobExecutionListener#stop(org.springframework.batch.core.domain.JobExecution)
*/
public void stop(JobExecution execution) {
// no-op
}
}

40
execution/src/main/java/org/springframework/batch/execution/facade/SimpleJobExecutorFacade.java
View File

@@ -30,7 +30,6 @@ import org.springframework.batch.core.configuration.NoSuchJobConfigurationExcept
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.execution.job.DefaultJobExecutor;
@@ -141,11 +140,16 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade,
JobInstance job = jobRepository.findOrCreateJob(jobConfiguration,
jobIdentifier);
JobExecution jobExecution = new JobExecution(job);
JobExecution execution = new JobExecution(job);
jobExecutor.run(jobConfiguration, jobExecution, this);
this.before(execution);
try {
jobExecutor.run(jobConfiguration, execution);
} finally {
this.after(execution);
}
return jobExecution.getExitStatus();
return execution.getExitStatus();
}
/**
@@ -167,6 +171,19 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade,
}
}
/**
* Broadcast stop signal to all the registered listeners.
*
* @param execution
*/
public void stop(JobExecution execution) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
JobExecutionListener listener = (JobExecutionListener) iterator
.next();
listener.stop(execution);
}
}
/**
* Internal accounting for the job execution. Callback at end of job
* delegating first to listeners, in reverse order to the list supplied, and
@@ -190,29 +207,32 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade,
}
}
/*
* (non-Javadoc)
/**
* Send a stop signal to all the running executions by setting their
* {@link RepeatContext} to terminate only. Then call the
* {@link JobExecutionListener#stop(JobExecution)} method.
*
* @see org.springframework.batch.container.BatchContainer#stop(org.springframework.batch.container.common.runtime.JobRuntimeInformation)
*/
public void stop(JobIdentifier runtimeInformation)
throws NoSuchJobExecutionException {
JobExecution jobExecutionContext = (JobExecution) jobExecutionRegistry
JobExecution execution = (JobExecution) jobExecutionRegistry
.get(runtimeInformation);
if (jobExecutionContext == null) {
if (execution == null) {
throw new NoSuchJobExecutionException("No such Job is executing: ["
+ runtimeInformation + "]");
}
for (Iterator iter = jobExecutionContext.getStepContexts().iterator(); iter
for (Iterator iter = execution.getStepContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setTerminateOnly();
}
for (Iterator iter = jobExecutionContext.getChunkContexts().iterator(); iter
for (Iterator iter = execution.getChunkContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setTerminateOnly();
}
this.stop(execution);
}
/**

4
execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java
View File

@@ -28,8 +28,8 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.executor.AbstractJobExecutor;
import org.springframework.batch.core.executor.ExitCodeExceptionClassifier;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.executor.StepExecutor;
import org.springframework.batch.core.executor.StepExecutorFactory;
import org.springframework.batch.core.executor.StepInterruptedException;
@@ -47,7 +47,7 @@ import org.springframework.batch.repeat.RepeatContext;
* @author Lucas Ward
* @author Dave Syer
*/
public class DefaultJobExecutor extends AbstractJobExecutor {
public class DefaultJobExecutor implements JobExecutor {
private static final SimpleStepExecutorFactory DEFAULT_STEP_EXECUTOR_FACTORY = new SimpleStepExecutorFactory();

118
execution/src/test/java/org/springframework/batch/execution/bootstrap/InterruptJobTests.java Normal file
View File

@@ -0,0 +1,118 @@
/*
* 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.execution.bootstrap;
import java.util.Collections;
import junit.framework.TestCase;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.execution.bootstrap.support.ThreadInterruptJobExecutionListener;
import org.springframework.batch.execution.configuration.MapJobConfigurationRegistry;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.execution.facade.SimpleJobExecutorFacade;
import org.springframework.batch.execution.repository.SimpleJobRepository;
import org.springframework.batch.execution.repository.dao.MapJobDao;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
/**
* @author Dave Syer
*
*/
public class InterruptJobTests extends TestCase {
public void testInterruptUsingListener() throws Exception {
// final InterruptibleFacade facade = new InterruptibleFacade();
// facade.setListener(new ThreadInterruptJobExecutionListener());
final SimpleJobExecutorFacade facade = new SimpleJobExecutorFacade();
facade.setJobExecutor(new InterruptibleJobExecutor());
facade.setJobRepository(new SimpleJobRepository(new MapJobDao(),
new MapStepDao()));
facade.setJobExecutionListeners(Collections
.singletonList(new ThreadInterruptJobExecutionListener()));
MapJobConfigurationRegistry registry = new MapJobConfigurationRegistry();
facade.setJobConfigurationLocator(registry);
registry.register(new JobConfiguration("foo"));
final SimpleJobIdentifier identifier = new SimpleJobIdentifier("foo");
TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
Runnable launcherRunnable = new Runnable() {
public void run() {
try {
facade.start(identifier);
} catch (NoSuchJobConfigurationException e) {
fail("Unexpected NoSuchJobConfigurationException");
}
}
};
taskExecutor.execute(launcherRunnable);
// give the thread a second to start up
Thread.sleep(100);
assertTrue(facade.isRunning());
facade.stop(identifier);
Thread.sleep(100);
assertFalse(facade.isRunning());
}
/**
* Simple {@link JobExecutorFacade} that can be used to test thread
* interruption. Mimics the implementation of the
* {@link SimpleJobExecutorFacade} with the use of a listener, but silently
* allows the current thread to be interrupted.
*
* @author Dave Syer
*
*/
private class InterruptibleJobExecutor implements JobExecutor {
/*
* (non-Javadoc)
*
* @see org.springframework.batch.core.executor.JobExecutor#run(org.springframework.batch.core.configuration.JobConfiguration,
* org.springframework.batch.core.domain.JobExecution)
*/
public ExitStatus run(JobConfiguration configuration,
JobExecution execution) throws BatchCriticalException {
try {
// 1 seconds should be long enough to allow the thread to be
// run and for interrupt to be called;
Thread.sleep(3000);
return ExitStatus.FAILED;
} catch (InterruptedException ex) {
// thread interrupted, allow to exit normally
return ExitStatus.FAILED;
}
}
}
}

66
execution/src/test/java/org/springframework/batch/execution/bootstrap/JobExecutionNotificationPublisherTests.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* 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.execution.bootstrap;
import java.util.ArrayList;
import java.util.List;
import javax.management.Notification;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.UnableToSendNotificationException;
import junit.framework.TestCase;
/**
* @author Dave Syer
*
*/
public class JobExecutionNotificationPublisherTests extends TestCase {
JobExecutionNotificationPublisher publisher = new JobExecutionNotificationPublisher();
public void testRepeatOperationsBeforeNotUsed() throws Exception {
final List list = new ArrayList();
publisher.setNotificationPublisher(new NotificationPublisher() {
public void sendNotification(Notification notification)
throws UnableToSendNotificationException {
list.add(notification);
}
});
publisher.onApplicationEvent(new RepeatOperationsApplicationEvent(this,
"foo", RepeatOperationsApplicationEvent.BEFORE) {
});
assertEquals(0, list.size());
}
public void testRepeatOperationsOpenUsed() throws Exception {
final List list = new ArrayList();
publisher.setNotificationPublisher(new NotificationPublisher() {
public void sendNotification(Notification notification)
throws UnableToSendNotificationException {
list.add(notification);
}
});
publisher.onApplicationEvent(new RepeatOperationsApplicationEvent(this,
"foo", RepeatOperationsApplicationEvent.OPEN));
assertEquals(1, list.size());
assertEquals("foo", ((Notification) list.get(0)).getMessage()
.substring(0, 3));
}
}

66
execution/src/test/java/org/springframework/batch/execution/bootstrap/SimpleJobLauncherTests.java
View File

@@ -21,12 +21,10 @@ import junit.framework.TestCase;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.runtime.SimpleJobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
public class SimpleJobLauncherTests extends TestCase {
@@ -35,8 +33,7 @@ public class SimpleJobLauncherTests extends TestCase {
try {
launcher.afterPropertiesSet();
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {
} catch (IllegalArgumentException e) {
// expected
assertTrue(e.getMessage().indexOf("required") >= 0);
}
@@ -48,9 +45,10 @@ public class SimpleJobLauncherTests extends TestCase {
launcher.run();
// should do nothing
fail("Expected NoSuchJobConfigurationException");
}
catch (NoSuchJobConfigurationException e) {
assertTrue("Message should mention null job name: "+e.getMessage(), e.getMessage().toLowerCase().indexOf("null")>=0);
} catch (NoSuchJobConfigurationException e) {
assertTrue("Message should mention null job name: "
+ e.getMessage(), e.getMessage().toLowerCase().indexOf(
"null") >= 0);
}
}
@@ -68,44 +66,13 @@ public class SimpleJobLauncherTests extends TestCase {
assertFalse(launcher.isRunning());
}
public void testInterruptContainer() throws Exception {
final SimpleJobLauncher launcher = new SimpleJobLauncher();
launcher.setJobIdentifierFactory(new SimpleJobIdentifierFactory());
InterruptibleFacade jobExecutorFacade = new InterruptibleFacade();
launcher.setJobExecutorFacade(jobExecutorFacade);
launcher.setJobConfigurationName(new JobConfiguration("foo").getName());
TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
Runnable launcherRunnable = new Runnable() {
public void run() {
try {
launcher.run();
} catch (NoSuchJobConfigurationException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
taskExecutor.execute(launcherRunnable);
// give the thread a second to start up
Thread.sleep(100);
assertTrue(launcher.isRunning());
launcher.stop();
Thread.sleep(100);
assertFalse(launcher.isRunning());
}
public void testStopOnUnranLauncher() {
public void testStopOnNotRunningLauncher() {
SimpleJobLauncher launcher = new SimpleJobLauncher();
assertFalse(launcher.isRunning());
// no exception should be thrown if stop is called on unran
// container
// this is to fullfill the contract outlined in Lifecycle#stop().
// no exception should be thrown if stop is called on
// a launcher that is not running.
launcher.stop();
}
@@ -113,6 +80,7 @@ public class SimpleJobLauncherTests extends TestCase {
/*
* (non-Javadoc)
*
* @see org.springframework.batch.container.BatchContainer#run()
*/
public void run() {
@@ -120,21 +88,19 @@ public class SimpleJobLauncherTests extends TestCase {
// 1 seconds should be long enough to allow the thread to be
// run and for interrupt to be called;
Thread.sleep(300);
//return ExitStatus.FAILED;
}
catch (InterruptedException ex) {
// return ExitStatus.FAILED;
} catch (InterruptedException ex) {
// thread interrupted, allow to exit normally
//return ExitStatus.FAILED;
// return ExitStatus.FAILED;
}
}
public ExitStatus start(JobIdentifier runtimeInformation) {
run();
return ExitStatus.FAILED;
}
public ExitStatus start(JobIdentifier jobIdentifier,
JobExecutionListener listener)
throws NoSuchJobConfigurationException {
@@ -144,7 +110,7 @@ public class SimpleJobLauncherTests extends TestCase {
public void stop(JobIdentifier runtimeInformation) {
// not needed
}
public boolean isRunning() {
// not needed
return false;

38
execution/src/test/java/org/springframework/batch/execution/bootstrap/TaskExecutorJobLauncherTests.java
View File

@@ -20,31 +20,26 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import javax.management.Notification;
import junit.framework.TestCase;
import org.easymock.MockControl;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.core.runtime.SimpleJobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.UnableToSendNotificationException;
public class TaskExecutorJobLauncherTests extends TestCase {
private TaskExecutorJobLauncher launcher = new TaskExecutorJobLauncher();
private SimpleJobLauncher launcher = new SimpleJobLauncher();
protected void setUp() throws Exception {
super.setUp();
@@ -77,35 +72,6 @@ public class TaskExecutorJobLauncherTests extends TestCase {
// nothing happens
}
public void testRepeatOperationsBeforeNotUsed() throws Exception {
final List list = new ArrayList();
launcher.setNotificationPublisher(new NotificationPublisher() {
public void sendNotification(Notification notification)
throws UnableToSendNotificationException {
list.add(notification);
}
});
launcher.onApplicationEvent(new RepeatOperationsApplicationEvent(this,
"foo", RepeatOperationsApplicationEvent.BEFORE) {
});
assertEquals(0, list.size());
}
public void testRepeatOperationsOpenUsed() throws Exception {
final List list = new ArrayList();
launcher.setNotificationPublisher(new NotificationPublisher() {
public void sendNotification(Notification notification)
throws UnableToSendNotificationException {
list.add(notification);
}
});
launcher.onApplicationEvent(new RepeatOperationsApplicationEvent(this,
"foo", RepeatOperationsApplicationEvent.OPEN));
assertEquals(1, list.size());
assertEquals("foo", ((Notification) list.get(0)).getMessage()
.substring(0, 3));
}
public void testStatisticsRetrieved() throws Exception {
MockControl control = MockControl
.createControl(JobExecutorFacadeWithStatistics.class);

82
execution/src/test/java/org/springframework/batch/execution/facade/JobExecutionListenerSupportTests.java Normal file
View File

@@ -0,0 +1,82 @@
/*
* 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.execution.facade;
import java.util.ArrayList;
import java.util.List;
import junit.framework.TestCase;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutionListenerSupport;
/**
* @author Dave Syer
*
*/
public class JobExecutionListenerSupportTests extends TestCase {
private List list = new ArrayList();
/**
* Test method for
* {@link org.springframework.batch.execution.facade.JobExecutionListenerSupport#after(org.springframework.batch.core.domain.JobExecution)}.
*/
public void testAfter() {
JobExecutionListener listener = new JobExecutionListenerSupport() {
public void after(JobExecution execution) {
super.after(execution);
list.add("after");
}
};
listener.after(null);
assertEquals(1, list.size());
}
/**
* Test method for
* {@link org.springframework.batch.execution.facade.JobExecutionListenerSupport#before(org.springframework.batch.core.domain.JobExecution)}.
*/
public void testBefore() {
JobExecutionListener listener = new JobExecutionListenerSupport() {
public void before(JobExecution execution) {
super.before(execution);
list.add("after");
}
};
listener.before(null);
assertEquals(1, list.size());
}
/**
* Test method for
* {@link org.springframework.batch.execution.facade.JobExecutionListenerSupport#before(org.springframework.batch.core.domain.JobExecution)}.
*/
public void testStop() {
JobExecutionListener listener = new JobExecutionListenerSupport() {
public void stop(JobExecution execution) {
super.stop(execution);
list.add("stop");
}
};
listener.stop(null);
assertEquals(1, list.size());
}
}

34
execution/src/test/java/org/springframework/batch/execution/facade/SimpleJobExecutorFacadeTests.java
View File

@@ -30,8 +30,6 @@ import org.springframework.batch.core.configuration.JobConfigurationLocator;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.executor.AbstractJobExecutor;
import org.springframework.batch.core.executor.JobExecutionListenerSupport;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
@@ -90,7 +88,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
private JobInstance setUpFacadeForNormalStart() {
jobIdentifier = new SimpleJobIdentifier("bar");
jobRepository.findOrCreateJob(jobConfiguration, jobIdentifier);
jobExecutor = new AbstractJobExecutor() {
jobExecutor = new JobExecutor() {
public ExitStatus run(JobConfiguration configuration,
JobExecution execution)
throws BatchCriticalException {
@@ -114,7 +112,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
}
public void testIsRunning() throws Exception {
jobExecutorFacade.setJobExecutor(new AbstractJobExecutor() {
jobExecutorFacade.setJobExecutor(new JobExecutor() {
public ExitStatus run(JobConfiguration configuration,
JobExecution execution)
throws BatchCriticalException {
@@ -193,6 +191,15 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
"TestJob");
JobExecution execution = new JobExecution(new JobInstance(
runtimeInformation, new Long(0)));
List listeners = new ArrayList();
listeners.add(new JobExecutionListenerSupport() {
public void stop(JobExecution execution) {
list.add("one");
}
});
jobExecutorFacade.setJobExecutionListeners(listeners);
registerExecution(runtimeInformation, execution);
RepeatContextSupport stepContext = new RepeatContextSupport(null);
@@ -204,6 +211,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertTrue(stepContext.isCompleteOnly());
assertTrue(chunkContext.isCompleteOnly());
assertEquals(1, list.size());
}
public void testStatisticsWithNoContext() throws Exception {
@@ -222,6 +230,24 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertTrue(statistics.containsKey("job1.step1"));
}
public void testListenersCalledLastOnStop() throws Exception {
List listeners = new ArrayList();
listeners.add(new JobExecutionListenerSupport() {
public void stop(JobExecution execution) {
list.add("one");
}
});
listeners.add(new JobExecutionListenerSupport() {
public void stop(JobExecution execution) {
list.add("two");
}
});
jobExecutorFacade.setJobExecutionListeners(listeners);
jobExecutorFacade.stop(jobExecution);
assertEquals(2, list.size());
assertEquals("two", list.get(1));
}
public void testListenersCalledLastOnAfter() throws Exception {
List listeners = new ArrayList();
listeners.add(new JobExecutionListenerSupport() {