Initial SF/SI 4

More

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java

@@ -18,11 +18,11 @@ package org.springframework.batch.integration.async;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.scope.context.StepContext;
 import org.springframework.batch.core.scope.context.StepSynchronizationManager;
-import org.springframework.integration.Message;
-import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.ChannelInterceptor;
 import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * A {@link ChannelInterceptor} that adds the current {@link StepExecution} (if
@@ -30,9 +30,9 @@ import org.springframework.integration.support.MessageBuilder;
  * can then take advantage of the step context without needing to be step
  * scoped, which is a problem for handlers executing in another thread because
  * the scope context is not available.
- * 
+ *
  * @author Dave Syer
- * 
+ *
  */
 public class StepExecutionInterceptor extends ChannelInterceptorAdapter {
 

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

@@ -34,10 +34,10 @@ import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.integration.Message;
-import org.springframework.integration.core.MessagingOperations;
-import org.springframework.integration.core.PollableChannel;
-import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 
 public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>,
@@ -51,13 +51,13 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	private static final long DEFAULT_THROTTLE_LIMIT = 6;
 
-	private MessagingOperations messagingGateway;
+	private MessagingTemplate messagingGateway;
 
-	private LocalState localState = new LocalState();
+	private final LocalState localState = new LocalState();
 
 	private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
 
-	private int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
+	private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
 
 	private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
 
@@ -67,7 +67,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	 * The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
 	 * multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
 	 * slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
-	 * 
+	 *
 	 * @param maxWaitTimeouts the maximum number of wait timeouts
 	 */
 	public void setMaxWaitTimeouts(int maxWaitTimeouts) {
@@ -83,7 +83,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		this.throttleLimit = throttleLimit;
 	}
 
-	public void setMessagingOperations(MessagingOperations messagingGateway) {
+	public void setMessagingOperations(MessagingTemplate messagingGateway) {
 		this.messagingGateway = messagingGateway;
 	}
 
@@ -184,7 +184,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	/**
 	 * Wait until all the results that are in the pipeline come back to the reply channel.
-	 * 
+	 *
 	 * @return true if successfully received a result, false if timed out
 	 */
 	private boolean waitForResults() throws AsynchronousFailureException {
@@ -210,9 +210,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	/**
 	 * Get the next result if it is available (within the timeout specified in the gateway), otherwise do nothing.
-	 * 
+	 *
 	 * @throws AsynchronousFailureException If there is a response and it contains a failed chunk response.
-	 * 
+	 *
 	 * @throws IllegalStateException if the result contains the wrong job instance id (maybe we are sharing a channel
 	 * and we shouldn't be)
 	 */
@@ -263,17 +263,17 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	private static class LocalState {
 
-		private AtomicInteger current = new AtomicInteger(-1);
+		private final AtomicInteger current = new AtomicInteger(-1);
 
-		private AtomicInteger actual = new AtomicInteger();
+		private final AtomicInteger actual = new AtomicInteger();
 
-		private AtomicInteger expected = new AtomicInteger();
+		private final AtomicInteger expected = new AtomicInteger();
 
-		private AtomicInteger redelivered = new AtomicInteger();
+		private final AtomicInteger redelivered = new AtomicInteger();
 
 		private StepExecution stepExecution;
 
-		private Queue<ChunkResponse> contributions = new LinkedBlockingQueue<ChunkResponse>();
+		private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<ChunkResponse>();
 
 		public int getExpecting() {
 			return expected.get() - actual.get();

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

@@ -3,11 +3,11 @@ package org.springframework.batch.integration.chunk;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.InitializingBean;
-import org.springframework.integration.Message;
-import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.ChannelInterceptor;
 import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 
@@ -16,9 +16,9 @@ import org.springframework.util.Assert;
  * <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 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
- * 
+ *
  */
 public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter implements InitializingBean {
 
@@ -44,7 +44,7 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
 	/**
 	 * Optional MessageChannel for injecting the message received from the source (defaults to the channel intercepted
 	 * in {@link #preReceive(MessageChannel)}).
-	 * 
+	 *
 	 * @param channel the channel to set
 	 */
 	public void setChannel(MessageChannel channel) {
@@ -69,7 +69,7 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
 	/**
 	 * Receive from the {@link MessageSource} and send immediately to the input channel, so that the call that we are
 	 * intercepting always a message to receive.
-	 * 
+	 *
 	 * @see ChannelInterceptorAdapter#preReceive(MessageChannel)
 	 */
 	@Override

spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingGateway.java

@@ -19,9 +19,9 @@ package org.springframework.batch.integration.launch;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobExecutionException;
 import org.springframework.batch.core.launch.JobLauncher;
-import org.springframework.integration.Message;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 /**

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java

@@ -11,15 +11,15 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.partition.PartitionHandler;
 import org.springframework.batch.core.partition.StepExecutionSplitter;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.integration.Message;
-import org.springframework.integration.MessageChannel;
 import org.springframework.integration.annotation.Aggregator;
 import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.Payloads;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.core.MessagingOperations;
-import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -30,10 +30,10 @@ import org.springframework.util.Assert;
  * well as a remote web service or JMS implementation. If a remote worker fails or doesn't send a reply message, the job
  * will fail and can be restarted to pick up missing messages and processing. The remote workers need access to the
  * Spring Batch {@link JobRepository} so that the shared state across those restarts can be managed centrally.
- * 
+ *
  * @author Dave Syer
  * @author Will Schipp
- * 
+ *
  */
 @MessageEndpoint
 public class MessageChannelPartitionHandler implements PartitionHandler {
@@ -42,10 +42,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 
 	private int gridSize = 1;
 
-	private MessagingOperations messagingGateway;
+	private MessagingTemplate messagingGateway;
 
 	private String stepName;
-	
+
 	/**
 	 * pollable channel for the replies
 	 */
@@ -62,10 +62,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 	 * internally: <ul> <li>request channel capable of accepting {@link StepExecutionRequest} payloads</li> <li>reply
 	 * channel that returns a list of {@link StepExecution} results</li> </ul> The timeout for the repoy should be set
 	 * sufficiently long that the remote steps have time to complete.
-	 * 
+	 *
 	 * @param messagingGateway the {@link MessagingOperations} to set
 	 */
-	public void setMessagingOperations(MessagingOperations messagingGateway) {
+	public void setMessagingOperations(MessagingTemplate messagingGateway) {
 		this.messagingGateway = messagingGateway;
 	}
 
@@ -73,7 +73,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 	 * Passed to the {@link StepExecutionSplitter} in the {@link #handle(StepExecutionSplitter, StepExecution)} method,
 	 * instructing it how many {@link StepExecution} instances are required, ideally. The {@link StepExecutionSplitter}
 	 * is allowed to ignore the grid size in the case of a restart, since the input data partitions must be preserved.
-	 * 
+	 *
 	 * @param gridSize the number of step executions that will be created
 	 */
 	public void setGridSize(int gridSize) {
@@ -85,7 +85,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 	 * regular Spring Batch step, with all the business logic required to complete an execution based on the input
 	 * parameters in its {@link StepExecution} context. The name will be translated into a {@link Step} instance by the
 	 * remote worker.
-	 * 
+	 *
 	 * @param stepName the name of the {@link Step} instance to execute business logic
 	 */
 	public void setStepName(String stepName) {
@@ -110,7 +110,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 	 * receives the result back as a list of {@link StepExecution} on a reply channel. Use the {@link #aggregate(List)}
 	 * method as an aggregator of the individual remote replies. The receive timeout needs to be set realistically in
 	 * the {@link MessagingOperations} <b>and</b> the aggregator, so that there is a good chance of all work being done.
-	 * 
+	 *
 	 * @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
 	 */
 	public Collection<StepExecution> handle(StepExecutionSplitter stepExecutionSplitter,
@@ -118,11 +118,11 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 
 		Set<StepExecution> split = stepExecutionSplitter.split(masterStepExecution, gridSize);
 		int count = 0;
-		
+
 		if (replyChannel == null) {
 			replyChannel = new QueueChannel();
-		}//end if 
-		
+		}//end if
+
 		for (StepExecution stepExecution : split) {
 			Message<StepExecutionRequest> request = createMessage(count++, split.size(), new StepExecutionRequest(
 					stepName, stepExecution.getJobExecutionId(), stepExecution.getId()), replyChannel);