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

Update for SI latest interface changes. Transactional stuff still broken, but now it compiles.

Browse Source
This commit is contained in:
dsyer
2008-08-01 06:57:18 +00:00
parent ccb2467cce
commit 96cbf19f2e
9 changed files with 64 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -16,9 +16,10 @@ import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
@@ -37,9 +38,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private static final long DEFAULT_THROTTLE_LIMIT = 6;
private MessageChannel requestChannel;
private MessageTarget target;
private MessageChannel replyChannel;
private BlockingSource<ChunkResponse> source;
// TODO: abstract the state or make a factory for this writer?
private LocalState localState = new LocalState();
@@ -55,12 +56,12 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
this.throttleLimit = throttleLimit;
}
public void setReplyChannel(MessageChannel replyChannel) {
this.replyChannel = replyChannel;
public void setSource(BlockingSource<ChunkResponse> source) {
this.source = source;
}
public void setRequestChannel(MessageChannel requestChannel) {
this.requestChannel = requestChannel;
public void setTarget(MessageTarget target) {
this.target = target;
}
public void write(T item) throws Exception {
@@ -95,7 +96,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
logger.debug("Dispatching chunk: " + processed);
ChunkRequest<T> request = new ChunkRequest<T>(processed, localState.getJobId(), localState.getSkipCount());
GenericMessage<ChunkRequest<T>> message = new GenericMessage<ChunkRequest<T>>(request);
requestChannel.send(message);
target.send(message);
localState.expected++;
}
@@ -176,9 +177,10 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
* otherwise return null.
*/
private void getNextResult(long timeout) {
Message<?> message = replyChannel.receive(timeout);
// TODO: use the timeout
Message<ChunkResponse> message = source.receive(timeout);
if (message != null) {
ChunkResponse payload = (ChunkResponse) message.getPayload();
ChunkResponse payload = message.getPayload();
Long jobInstanceId = payload.getJobId();
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["

31
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java
View File

@@ -23,9 +23,10 @@ import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
@@ -39,9 +40,9 @@ public class MessageOrientedStep extends AbstractStep {
*/
public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING";
private MessageChannel requestChannel;
private MessageTarget target;
private MessageChannel replyChannel;
private BlockingSource<JobExecutionRequest> source;
private static int MINUTE = 1000 * 60;
@@ -75,21 +76,21 @@ public class MessageOrientedStep extends AbstractStep {
}
/**
* Public setter for the requestChannel.
* @param requestChannel the requestChannel to set
* Public setter for the target.
* @param target the target to set
*/
@Required
public void setRequestChannel(MessageChannel requestChannel) {
this.requestChannel = requestChannel;
public void setTarget(MessageTarget target) {
this.target = target;
}
/**
* Public setter for the replyChannel.
* @param replyChannel the replyChannel to set
* Public setter for the source.
* @param source the source to set
*/
@Required
public void setReplyChannel(MessageChannel replyChannel) {
this.replyChannel = replyChannel;
public void setSource(BlockingSource<JobExecutionRequest> source) {
this.source = source;
}
/*
@@ -112,7 +113,7 @@ public class MessageOrientedStep extends AbstractStep {
executionContext.putString(WAITING, "true");
// TODO: need these two lines to be atomic
getJobRepository().update(stepExecution);
requestChannel.send(new GenericMessage<JobExecutionRequest>(request));
target.send(new GenericMessage<JobExecutionRequest>(request));
waitForReply(request.getJobId());
}
@@ -128,14 +129,14 @@ public class MessageOrientedStep extends AbstractStep {
long maxCount = executionTimeout / timeout;
long count = 0;
// TODO: use a ReponseCorrelator?, or just a SynchronousChannel
while (count++ < maxCount) {
Message<?> message = replyChannel.receive(timeout);
// TODO: timeout?
Message<JobExecutionRequest> message = source.receive(timeout);
if (message != null) {
JobExecutionRequest payload = (JobExecutionRequest) message.getPayload();
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 ["

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java
View File

@@ -10,6 +10,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.test.context.ContextConfiguration;
@@ -26,7 +27,7 @@ public class SmokeTests {
@Autowired
@Qualifier("smokeout")
private MessageChannel smokeout;
private BlockingSource<String> smokeout;
// This has to be static because the MessageBus registers the handler
// more than once (every time a test instance is created), but only one of
@@ -48,7 +49,7 @@ public class SmokeTests {
@Test
public void testVanillaSendAndReceive() throws Exception {
smokein.send(new GenericMessage<String>("foo"));
Message<?> message = smokeout.receive(100);
Message<String> message = smokeout.receive(100);
String result = (String) (message == null ? null : message.getPayload());
assertEquals("foo: 1", result);
assertEquals(1, count);

11
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
View File

@@ -32,7 +32,7 @@ import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.test.context.ContextConfiguration;
@@ -47,11 +47,11 @@ public class ChunkMessageItemWriterIntegrationTests {
@Autowired
@Qualifier("requests")
private MessageChannel requests;
private PollableChannel requests;
@Autowired
@Qualifier("replies")
private MessageChannel replies;
private PollableChannel replies;
private SimpleStepFactoryBean<Object> factory;
@@ -59,6 +59,7 @@ public class ChunkMessageItemWriterIntegrationTests {
private static long jobCounter;
@SuppressWarnings("unchecked")
@Before
public void setUp() {
@@ -71,8 +72,8 @@ public class ChunkMessageItemWriterIntegrationTests {
factory.setItemWriter(writer);
factory.setCommitInterval(4);
writer.setReplyChannel(replies);
writer.setRequestChannel(requests);
writer.setSource(replies);
writer.setTarget(requests);
TestItemWriter.count = 0;

3
spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java
View File

@@ -28,6 +28,7 @@ import org.springframework.core.io.Resource;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.test.context.ContextConfiguration;
@@ -48,7 +49,7 @@ public class ResourceSplitterIntegrationTests {
@Autowired
@Qualifier("requests")
private MessageChannel requests;
private PollableChannel requests;
/*
* This is so cool (but see INT-190)...<br/>

4
spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java
View File

@@ -22,7 +22,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -36,7 +36,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class MessageChannelItemWriterIntegrationTests {
@Autowired
private MessageChannel channel;
private PollableChannel channel;
@Autowired
private ItemWriter<String> itemWriter;

23
spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java
View File

@@ -33,11 +33,13 @@ import org.springframework.batch.core.StepExecution;
import org.springframework.batch.integration.JobRepositorySupport;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.ReflectionUtils;
@@ -53,15 +55,16 @@ public class MessageOrientedStepTests {
private DirectChannel requestChannel;
private MessageChannel replyChannel;
private PollableChannel replyChannel;
@SuppressWarnings("unchecked")
@Before
public void createStep() {
replyChannel = new ThreadLocalChannel();
requestChannel = new DirectChannel();
step.setName("step");
step.setRequestChannel(requestChannel);
step.setReplyChannel(replyChannel);
step.setTarget(requestChannel);
step.setSource(replyChannel);
step.setStartLimit(10);
step.setJobRepository(new JobRepositorySupport());
JobInstance jobInstance = new JobInstance(0L, new JobParameters(), "job");
@@ -70,12 +73,12 @@ public class MessageOrientedStepTests {
/**
* Test method for
* {@link org.springframework.batch.integration.job.MessageOrientedStep#setRequestChannel(org.springframework.integration.channel.MessageChannel)}.
* {@link org.springframework.batch.integration.job.MessageOrientedStep#setTarget(MessageTarget)}.
*/
@Test
public void testSetRequestChannel() {
Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setRequestChannel",
new Class<?>[] { MessageChannel.class });
Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setTarget",
new Class<?>[] { MessageTarget.class });
assertNotNull(method);
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
assertEquals(1, annotations.length);
@@ -84,12 +87,12 @@ public class MessageOrientedStepTests {
/**
* Test method for
* {@link org.springframework.batch.integration.job.MessageOrientedStep#setReplyChannel(org.springframework.integration.channel.MessageChannel)}.
* {@link org.springframework.batch.integration.job.MessageOrientedStep#setSource(MessageSource)}.
*/
@Test
public void testSetReplyChannel() {
Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setReplyChannel",
new Class<?>[] { MessageChannel.class });
Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource",
new Class<?>[] { BlockingSource.class });
assertNotNull(method);
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
assertEquals(1, annotations.length);

6
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java
View File

@@ -15,7 +15,7 @@ import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.integration.JobSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
@@ -30,11 +30,11 @@ public class JobLaunchingMessageHandlerIntegrationTests {
@Autowired
@Qualifier("requests")
private MessageChannel requestChannel;
private PollableChannel requestChannel;
@Autowired
@Qualifier("response")
private MessageChannel responseChannel;
private PollableChannel responseChannel;
private JobSupport job = new JobSupport("testJob");

11
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java
View File

@@ -37,6 +37,7 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
import org.springframework.integration.scheduling.TaskScheduler;
@@ -385,7 +386,7 @@ public class PollableSourceRetryTests {
* @return
*/
private DirectChannel getChannel(MessageTarget handler, MessageSource<Object> source) {
DirectChannel channel = new DirectChannel(source);
DirectChannel channel = new DirectChannel();
channel.setName("input");
channel.subscribe(handler);
return channel;
@@ -400,7 +401,7 @@ public class PollableSourceRetryTests {
lifecycle.stop();
}
private MessageSource<Object> getPollableSource(List<String> list) {
private PollableSource<Object> getPollableSource(List<String> list) {
final ItemReader<String> reader = new ListItemReader<String>(list) {
public String read() {
String item = super.read();
@@ -408,10 +409,12 @@ public class PollableSourceRetryTests {
return item;
}
};
MessageSource<Object> source = new MessageSource<Object>() {
PollableSource<Object> source = new PollableSource<Object>() {
public Message<Object> receive() {
try {
return new GenericMessage<Object>(reader.read());
String payload = reader.read();
if (payload==null) return null;
return new GenericMessage<Object>(payload);
}
catch (RuntimeException e) {
throw e;