BATCHADM-48: replace MessageOrientedStep
This commit is contained in:
David Syer
committed by
Michael Minella
Michael Minella
parent
ed926ea7b8
commit
a775465c67
@@ -1,24 +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.integration.job;
|
||||
|
||||
/**
|
||||
* @author dsyer
|
||||
*
|
||||
*/
|
||||
public class JobExecutionReply {
|
||||
|
||||
}
|
||||
@@ -1,114 +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.integration.job;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
|
||||
/**
|
||||
* Encapsulation of a request to execute a job execution through a message flow
|
||||
* consisting of step handlers. A handler should pass the message on as it is,
|
||||
* modifying the request properties as necessary. Generally a handler will
|
||||
* execute a step as part of the {@link JobExecution} passed in, and should
|
||||
* change the status to {@link BatchStatus#COMPLETED} if the step is successful
|
||||
* (generally a handler cannot determine if the whole job execution is complete,
|
||||
* so this is just information about the step).<br/>
|
||||
*
|
||||
* If the incoming status is {@link BatchStatus#FAILED},
|
||||
* {@link BatchStatus#ABANDONED} or {@link BatchStatus#STOPPING} the request
|
||||
* should be ignored by handlers (passed on without modification).
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class JobExecutionRequest {
|
||||
|
||||
private JobExecution jobExecution;
|
||||
|
||||
private BatchStatus status;
|
||||
|
||||
private Throwable throwable;
|
||||
|
||||
/**
|
||||
* @param jobExecution
|
||||
*/
|
||||
public JobExecutionRequest(JobExecution jobExecution) {
|
||||
this.jobExecution = jobExecution;
|
||||
status = jobExecution.getStatus();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the current job execution id
|
||||
*/
|
||||
public Long getJobId() {
|
||||
return this.jobExecution.getJobId();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the current {@link BatchStatus}
|
||||
*/
|
||||
public BatchStatus getStatus() {
|
||||
return status;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the status.
|
||||
* @param status the status to set
|
||||
*/
|
||||
public void setStatus(BatchStatus status) {
|
||||
this.status = status;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true if there are errors
|
||||
*/
|
||||
public boolean hasErrors() {
|
||||
return throwable != null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public getter for the throwable.
|
||||
* @return the throwable
|
||||
*/
|
||||
public Throwable getLastThrowable() {
|
||||
return throwable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the throwable.
|
||||
* @param throwable the throwable to set
|
||||
*/
|
||||
public void registerThrowable(Throwable throwable) {
|
||||
this.throwable = throwable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public getter for the jobExecution.
|
||||
* @return the jobExecution
|
||||
*/
|
||||
public JobExecution getJobExecution() {
|
||||
return jobExecution;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.lang.Object#toString()
|
||||
*/
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName()+": "+jobExecution;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,175 +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.integration.job;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.ExitStatus;
|
||||
import org.springframework.batch.core.JobInterruptedException;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.UnexpectedJobExecutionException;
|
||||
import org.springframework.batch.core.step.AbstractStep;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.beans.factory.annotation.Required;
|
||||
import org.springframework.integration.channel.PollableChannel;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.integration.message.GenericMessage;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class MessageOrientedStep extends AbstractStep {
|
||||
|
||||
/**
|
||||
* Key in execution context for flag to say we are waiting.
|
||||
*/
|
||||
public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING";
|
||||
|
||||
private MessageChannel outputChannel;
|
||||
|
||||
private PollableChannel source;
|
||||
|
||||
private static long MINUTE = 1000 * 60;
|
||||
|
||||
private long executionTimeout = 30*MINUTE ;
|
||||
|
||||
private long pollingInterval = 5;
|
||||
|
||||
/**
|
||||
* Public setter for the execution timeout in minutes. Defaults to 30.
|
||||
* @param executionTimeoutMinutes the timeout to set
|
||||
*/
|
||||
public void setExecutionTimeoutMinutes(int executionTimeoutMinutes) {
|
||||
this.executionTimeout = executionTimeoutMinutes * MINUTE;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the execution timeout in milliseconds. Defaults to 30 minutes.
|
||||
* @param executionTimeout
|
||||
*/
|
||||
public void setExecutionTimeout(long executionTimeout) {
|
||||
this.executionTimeout = executionTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the polling interval in milliseconds while waiting for
|
||||
* replies signalling the end of the step. Defaults to 5.
|
||||
* @param pollingInterval the polling interval to set
|
||||
*/
|
||||
public void setPollingInterval(long pollingInterval) {
|
||||
this.pollingInterval = pollingInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the target.
|
||||
* @param outputChannel the target to set
|
||||
*/
|
||||
@Required
|
||||
public void setOutputChannel(MessageChannel outputChannel) {
|
||||
this.outputChannel = outputChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the source.
|
||||
* @param source the source to set
|
||||
*/
|
||||
@Required
|
||||
public void setInputChannel(PollableChannel source) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see AbstractStep#execute(StepExecution)
|
||||
*/
|
||||
@Override
|
||||
protected void doExecute(StepExecution stepExecution) throws JobInterruptedException,
|
||||
UnexpectedJobExecutionException {
|
||||
|
||||
JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution());
|
||||
|
||||
ExecutionContext executionContext = stepExecution.getExecutionContext();
|
||||
|
||||
if (executionContext.containsKey(WAITING)) {
|
||||
// restart scenario: we are still waiting for a response
|
||||
waitForReply(request.getJobId());
|
||||
}
|
||||
else {
|
||||
executionContext.putString(WAITING, "true");
|
||||
getJobRepository().updateExecutionContext(stepExecution);
|
||||
outputChannel.send(new GenericMessage<JobExecutionRequest>(request));
|
||||
waitForReply(request.getJobId());
|
||||
}
|
||||
|
||||
stepExecution.setExitStatus(ExitStatus.COMPLETED);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param expectedJobId
|
||||
*/
|
||||
private void waitForReply(Long expectedJobId) {
|
||||
long timeout = pollingInterval;
|
||||
long maxCount = executionTimeout / timeout;
|
||||
long count = 0;
|
||||
|
||||
while (count++ < maxCount) {
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<JobExecutionRequest> message = (Message<JobExecutionRequest>) source.receive(timeout);
|
||||
|
||||
if (message != null) {
|
||||
|
||||
JobExecutionRequest payload = message.getPayload();
|
||||
Long jobInstanceId = payload.getJobId();
|
||||
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
|
||||
Assert.state(jobInstanceId.equals(expectedJobId), "Message contained wrong job instance id ["
|
||||
+ jobInstanceId + "] should have been [" + expectedJobId + "].");
|
||||
|
||||
if (payload.getStatus() == BatchStatus.COMPLETED) {
|
||||
// One of the steps decided we were finished
|
||||
// TODO: wait for all the other steps that might be
|
||||
// executing concurrently?
|
||||
break;
|
||||
}
|
||||
|
||||
if (payload.hasErrors()) {
|
||||
rethrow(payload.getLastThrowable());
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
if (count >= maxCount) {
|
||||
throw new StepExecutionTimeoutException("Timed out waiting for steps to execute.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param lastThrowable
|
||||
*/
|
||||
private static void rethrow(Throwable t) throws RuntimeException {
|
||||
if (t instanceof RuntimeException) {
|
||||
throw (RuntimeException) t;
|
||||
}
|
||||
if (t instanceof Exception) {
|
||||
throw new UnexpectedJobExecutionException("Unexpected checked exception thrown by step.", t);
|
||||
}
|
||||
throw (Error) t;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,107 +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.integration.job;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.job.SimpleStepHandler;
|
||||
import org.springframework.batch.core.job.StepHandler;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.beans.factory.annotation.Required;
|
||||
import org.springframework.integration.annotation.MessageEndpoint;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
@MessageEndpoint
|
||||
public class StepExecutionMessageHandler {
|
||||
|
||||
private Step step;
|
||||
|
||||
private StepHandler stepHandler;
|
||||
|
||||
/**
|
||||
* Public setter for the {@link Step}.
|
||||
* @param step the step to set
|
||||
*/
|
||||
@Required
|
||||
public void setStep(Step step) {
|
||||
this.step = step;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link JobRepository} that is needed to manage the
|
||||
* state of the batch meta domain (jobs, steps, executions) during the life
|
||||
* of a job.
|
||||
*
|
||||
* @param jobRepository
|
||||
*/
|
||||
@Required
|
||||
public void setJobRepository(JobRepository jobRepository) {
|
||||
stepHandler = new SimpleStepHandler(jobRepository);
|
||||
}
|
||||
|
||||
@ServiceActivator
|
||||
public JobExecutionRequest handle(JobExecutionRequest request) {
|
||||
|
||||
// Hand off immediately if the job has already failed
|
||||
if (isComplete(request)) {
|
||||
return request;
|
||||
}
|
||||
|
||||
JobExecution jobExecution = request.getJobExecution();
|
||||
|
||||
try {
|
||||
|
||||
stepHandler.handleStep(step, jobExecution);
|
||||
// (the job might actually not be complete, but the stage is).
|
||||
request.setStatus(BatchStatus.COMPLETED);
|
||||
|
||||
}
|
||||
catch (Exception e) {
|
||||
handleFailure(request, e);
|
||||
}
|
||||
catch (Error e) {
|
||||
handleFailure(request, e);
|
||||
throw e;
|
||||
}
|
||||
|
||||
return request;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param request
|
||||
* @return
|
||||
*/
|
||||
private boolean isComplete(JobExecutionRequest request) {
|
||||
return request.getStatus() == BatchStatus.FAILED || request.getStatus() == BatchStatus.ABANDONED
|
||||
|| request.getStatus() == BatchStatus.STOPPING;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param request
|
||||
* @param e
|
||||
*/
|
||||
private void handleFailure(JobExecutionRequest request, Throwable e) {
|
||||
request.registerThrowable(e);
|
||||
request.setStatus(BatchStatus.FAILED);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,36 +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.integration.job;
|
||||
|
||||
import org.springframework.batch.core.UnexpectedJobExecutionException;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class StepExecutionTimeoutException extends UnexpectedJobExecutionException {
|
||||
|
||||
/**
|
||||
* Constructs a new instance.
|
||||
*
|
||||
* @param msg the exception message.
|
||||
*
|
||||
*/
|
||||
public StepExecutionTimeoutException(String msg) {
|
||||
super(msg);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
/*
|
||||
* Copyright 2006-2010 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.integration.step;
|
||||
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.step.AbstractStep;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Provides a wrapper for an existing {@link Step}, delegating execution to it,
|
||||
* but serving all other operations locally.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class DelegateStep extends AbstractStep {
|
||||
|
||||
private Step delegate;
|
||||
|
||||
/**
|
||||
* @param delegate the delegate to set
|
||||
*/
|
||||
public void setDelegate(Step delegate) {
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check mandatory properties (delegate).
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert.state(delegate!=null, "A delegate Step must be provided");
|
||||
super.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doExecute(StepExecution stepExecution) throws Exception {
|
||||
delegate.execute(stepExecution);
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user