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

Upgrade Spring and add some JSON services

Browse Source
This commit is contained in:
David Syer
2010-07-25 12:44:28 +00:00
committed by Michael Minella
parent bcb78d8499
commit 6df8e4162b
5 changed files with 42 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java
View File

@@ -4,7 +4,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
@@ -14,8 +13,8 @@ import org.springframework.util.Assert;
/**
* A {@link ChannelInterceptor} that turns a pollable channel into a "pass-thru channel": if a client calls
* <code>receive()</code> on the channel it will delegate to a {@link MessageSource} to pull the message directly from
* an external source. This is particularly useful in combination with a {@link ThreadLocalChannel}, in which case the
* <code>receive()</code> can join a transaction which was started by the caller.
* an external source. This is particularly useful in combination with a message channel in thread scope, in which case
* the <code>receive()</code> can join a transaction which was started by the caller.
*
* @author Dave Syer
*
@@ -42,8 +41,8 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
}
/**
* Optional MessageChannel for injecting the message receieved from the source (defaults to the channel
* intercepted in {@link #preReceive(MessageChannel)}).
* Optional MessageChannel for injecting the message receieved from the source (defaults to the channel intercepted
* in {@link #preReceive(MessageChannel)}).
*
* @param channel the channel to set
*/
@@ -76,12 +75,12 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
public boolean preReceive(MessageChannel channel) {
Message<?> message = source.receive();
if (message != null) {
if (this.channel!=null) {
if (this.channel != null) {
channel = this.channel;
}
channel.send(message);
if (logger.isDebugEnabled()) {
logger.debug("Sent " + message + " to channel " + channel.getName());
logger.debug("Sent " + message + " to channel " + channel);
}
return true;
}

4
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptorTests.java
View File

@@ -4,7 +4,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.MessageSource;
@@ -33,7 +33,7 @@ public class MessageSourcePollerInterceptorTests {
@Test
public void testPreReceive() throws Exception {
MessageSourcePollerInterceptor interceptor = new MessageSourcePollerInterceptor(new TestMessageSource("foo"));
ThreadLocalChannel channel = new ThreadLocalChannel();
QueueChannel channel = new QueueChannel();
assertTrue(interceptor.preReceive(channel));
assertEquals("foo", channel.receive(10L).getPayload());
}

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

@@ -42,7 +42,10 @@ public class JobLaunchingMessageHandlerIntegrationTests {
@Before
public void setUp() {
responseChannel.purge(null);
Object message = "";
while (message!=null) {
message = responseChannel.receive(10L);
}
}
@Test