diff --git a/spring-batch-docs/asciidoc/spring-batch-integration.adoc b/spring-batch-docs/asciidoc/spring-batch-integration.adoc index 351f16aa1..f25cdda29 100644 --- a/spring-batch-docs/asciidoc/spring-batch-integration.adoc +++ b/spring-batch-docs/asciidoc/spring-batch-integration.adoc @@ -6,6 +6,8 @@ == Spring Batch Integration +include::toggle.adoc[] + [[spring-batch-integration-introduction]] === Spring Batch Integration Introduction @@ -38,10 +40,10 @@ provide methods to distribute workloads over a number of workers. This section covers the following key concepts: +[role="xmlContent"] * <> - - +[[continue-section-list]] * <> @@ -59,7 +61,7 @@ Batch Process Execution>> [[namespace-support]] - +[role="xmlContent"] ==== Namespace Support Since Spring Batch Integration 1.3, dedicated XML Namespace @@ -265,7 +267,8 @@ launch the job via the __Job Launching Gateway__, and then log the output of the `JobExecution` with the `logging-channel-adapter`. -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -292,6 +295,38 @@ launch the job via the __Job Launching Gateway__, and then log the output of the ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public FileMessageToJobRequest fileMessageToJobRequest() { + FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest(); + fileMessageToJobRequest.setFileParameterName("input.file.name"); + fileMessageToJobRequest.setJob(personJob()); + return fileMessageToJobRequest; +} + +@Bean +public JobLaunchingGateway jobLaunchingGateway() { + SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher(); + simpleJobLauncher.setJobRepository(jobRepository); + simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor()); + JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher); + + return jobLaunchingGateway; +} + +@Bean +public IntegrationFlow integrationFlow(JobLaunchingGateway jobLaunchingGateway) { + return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/myfiles")). + filter(new SimplePatternFileListFilter("*.csv")), + c -> c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1))). + handle(fileMessageToJobRequest()). + handle(jobLaunchingGateway). + log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload"). + get(); +} +---- [[example-itemreader-configuration]] @@ -303,7 +338,8 @@ configure our Spring Batch `ItemReader` (for example) to use the files found at the location defined by the job parameter called "input.file.name", as shown in the following bean configuration: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -312,6 +348,19 @@ by the job parameter called "input.file.name", as shown in the following bean co ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +@StepScope +public ItemReader sampleReader(@Value("#{jobParameters[input.file.name]}") String resource) { +... + FlatFileItemReader flatFileItemReader = new FlatFileItemReader(); + flatFileItemReader.setResource(new FileSystemResource(resource)); +... + return flatFileItemReader; +} +---- The main points of interest in the preceding example are injecting the value of `#{jobParameters['input.file.name']}` @@ -364,7 +413,9 @@ When this `Gateway` is receiving messages from a `PollableChannel`, you must either provide a global default `Poller` or provide a `Poller` sub-element to the `Job Launching Gateway`, as shown in the following example: -[source, xml] + +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -372,6 +423,18 @@ a global default `Poller` or provide a `Poller` sub-element to the ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +@ServiceActivator(inputChannel = "queueChannel", poller = @Poller(fixedRate="1000")) +public JobLaunchingGateway sampleJobLaunchingGateway() { + JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(jobLauncher()); + jobLaunchingGateway.setOutputChannel(replyChannel()); + return jobLaunchingGateway; +} +---- + [[providing-feedback-with-informational-messages]] ==== Providing Feedback with Informational Messages @@ -428,7 +491,8 @@ message to a `Gateway` for a First, create the notification integration beans: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -439,10 +503,31 @@ First, create the notification integration beans: ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +@ServiceActivator(inputChannel = "stepExecutionsChannel") +public LoggingHandler loggingHandler() { + LoggingHandler adapter = new LoggingHandler(LoggingHandler.Level.WARN); + adapter.setLoggerName("TEST_LOGGER"); + adapter.setLogExpressionString("headers.id + ': ' + payload"); + return adapter; +} + +@MessagingGateway(name = "notificationExecutionsListener", defaultRequestChannel = "stepExecutionsChannel") +public interface NotificationExecutionListener extends StepExecutionListener {} +---- + +[role="javaContent"] +NOTE: You will need to add the `@IntegrationComponentScan` annotation to your configuration. + +[[message-gateway-entry-list]] Second, modify your job to add a step-level listener: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -457,6 +542,18 @@ Second, modify your job to add a step-level listener: ---- +.Java Configuration +[source, java, role="javaContent"] +---- +public Job importPaymentsJob() { + return jobBuilderFactory.get("importPayments") + .start(stepBuilderFactory.get("step1") + .chunk(200) + .listener(notificationExecutionsListener()) + ... +} +---- + [[asynchronous-processors]] ==== Asynchronous Processors @@ -482,8 +579,8 @@ writes back the chunk as soon as all the results become available. The following example shows how to configuration the `AsyncItemProcessor`: - -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -496,6 +593,17 @@ The following example shows how to configuration the `AsyncItemProcessor`: ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public AsyncItemProcessor processor(ItemProcessor itemProcessor, TaskExecutor taskExecutor) { + AsyncItemProcessor asyncItemProcessor = new AsyncItemProcessor(); + asyncItemProcessor.setTaskExecutor(taskExecutor); + asyncItemProcessor.setDelegate(itemProcessor); + return asyncItemProcessor; +} +---- The `delegate` property refers to your `ItemProcessor` bean, and @@ -504,7 +612,8 @@ refers to the `TaskExecutor` of your choice. The following example shows how to configure the `AsyncItemWriter`: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -514,6 +623,16 @@ The following example shows how to configure the `AsyncItemWriter`: ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public AsyncItemWriter processor(ItemWriter itemWriter) { + AsyncItemWriter asyncItemWriter = new AsyncItemWriter(); + asyncItemWriter.setDelegate(itemWriter); + return asyncItemWriter; +} +---- Again, the `delegate` property is actually a reference to your `ItemWriter` bean. @@ -569,7 +688,8 @@ external systems for processing. A simple job with a step to be remotely chunked might have a configuration similar to the following: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -581,6 +701,20 @@ configuration similar to the following: ---- +.Java Configuration +[source, java, role="javaContent"] +---- +public Job chunkJob(ItemReader itemReader) { + return jobBuilderFactory.get("personJob") + .start(stepBuilderFactory.get("step1") + .chunk(200) + .reader(itemReader()) + .writer(itemWriter()) + .build()) + .build(); + } +---- + The `ItemReader` reference points to the bean you want to use for reading data on the master. The `ItemWriter` reference points to a special `ItemWriter` @@ -591,7 +725,8 @@ following configuration provides a basic master setup. You should check any additional component properties, such as throttle limits and so on, when implementing your use case. -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -627,6 +762,69 @@ throttle limits and so on, when implementing your use case. channel="replies"/> ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() { + ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(); + factory.setBrokerURL("tcp://localhost:61616"); + return factory; +} + +@Bean +public DirectChannel requests() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow jmsOutboundFlow() { + return IntegrationFlows.from("requests") + .handle(Jms.outboundGateway(connectionFactory()) + .requestDestination("requests")) + .get(); +} + +@Bean +public MessagingTemplate messagingTemplate() { + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(requests()); + template.setReceiveTimeout(2000); + return template; +} + +@Bean +@StepScope() +public ChunkMessageChannelItemWriter itemWriter() { + ChunkMessageChannelItemWriter chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter(); + chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate()); + chunkMessageChannelItemWriter.setReplyChannel(replies()); + return chunkMessageChannelItemWriter; +} + +@Bean +public RemoteChunkHandlerFactoryBean chunkHandler() { + RemoteChunkHandlerFactoryBean remoteChunkHandlerFactoryBean = new RemoteChunkHandlerFactoryBean(); + remoteChunkHandlerFactoryBean.setChunkWriter(itemWriter()); + remoteChunkHandlerFactoryBean.setStep(step1()); + return remoteChunkHandlerFactoryBean; +} + +@Bean +public QueueChannel replies() { + return new QueueChannel(); +} + +@Bean +public IntegrationFlow jmsReplies() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory()) + .configureListenerContainer(c -> c.subscriptionDurable(false)) + .destination("replies")) + .channel(replies()) + .get(); +} +---- The preceding configuration provides us with a number of beans. We configure our messaging middleware using ActiveMQ and the @@ -639,8 +837,8 @@ configured middleware. Now we can move on to the slave configuration, as shown in the following example: - -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -679,6 +877,60 @@ Now we can move on to the slave configuration, as shown in the following example ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() { + ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(); + factory.setBrokerURL("tcp://localhost:61616"); + return factory; +} + +@Bean +public DirectChannel requests() { + return new DirectChannel(); +} +@Bean +public DirectChannel replies() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow jmsIn() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory()) + .configureListenerContainer(c -> c.subscriptionDurable(false)) + .destination("requests")) + .channel(requests()) + .get(); +} + +@Bean +public IntegrationFlow outgoingReplies() { + return IntegrationFlows.from("replies") + .handle(Jms.outboundGateway(connectionFactory()) + .requestDestination("replies")) + .get(); +} + +@Bean +@ServiceActivator(inputChannel = "requests") +public AggregatorFactoryBean serviceActivator() throws Exception{ + AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); + aggregatorFactoryBean.setProcessorBean(chunkProcessorChunkHandler()); + aggregatorFactoryBean.setOutputChannel(replies()); + ... + return aggregatorFactoryBean; +} + +@Bean +public ChunkProcessorChunkHandler chunkProcessorChunkHandler() { + ChunkProcessorChunkHandler chunkProcessorChunkHandler = new ChunkProcessorChunkHandler(); + chunkProcessorChunkHandler.setChunkProcessor(new SimpleChunkProcessor(personItemProcessor(), personItemWriter())); + return chunkProcessorChunkHandler; +} +---- Most of these configuration items should look familiar from the master configuration. Slaves do not need access to @@ -744,8 +996,8 @@ the `MessageChannelPartitionHandler` and JMS configuration: - -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -796,9 +1048,111 @@ configuration: class="org.springframework.batch.integration.partition.BeanFactoryStepLocator" /> ---- +.Java Configuration +[source, java, role="javaContent"] +---- +@Bean +public PartitionHandler partitionHandler() { + MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); + partitionHandler.setStepName("step1"); + partitionHandler.setGridSize(3); + partitionHandler.setReplyChannel(outboundReplies()); + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(outboundRequests()); + template.setReceiveTimeout(100000); + partitionHandler.setMessagingOperations(template); + return partitionHandler; +} + +@Bean +public DirectChannel outboundRequests() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow outboundJmsRequests() { + return IntegrationFlows.from("outboundRequests") + .handle(Jms.outboundGateway(connectionFactory()) + .requestDestination("requestsQueue")) + .get(); +} + +@Bean +public DirectChannel inboundRequests() { + return new DirectChannel(); +} + +public IntegrationFlow inboundJmsRequests() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory()) + .configureListenerContainer(c -> c.subscriptionDurable(false)) + .destination("requestsQueue")) + .channel(inboundRequests()) + .get(); + } + +@Bean +public StepExecutionRequestHandler stepExecutionRequestHandler() { + StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); + stepExecutionRequestHandler.setJobExplorer(jobExplorer); + stepExecutionRequestHandler.setStepLocator(stepLocator()); + return stepExecutionRequestHandler; +} + +@Bean +@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging") +public StepExecutionRequestHandler serviceActivator() throws Exception { + return stepExecutionRequestHandler(); +} + +@Bean +public DirectChannel outboundStaging() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow outboundJmsStaging() { + return IntegrationFlows.from("outboundStaging") + .handle(Jms.outboundGateway(connectionFactory()) + .requestDestination("stagingQueue")) + .get(); +} + +@Bean +public DirectChannel inboundStaging() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow inboundJmsStaging() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory()) + .configureListenerContainer(c -> c.subscriptionDurable(false)) + .destination("stagingQueue")) + .channel(inboundStaging()) + .get(); +} + +@Bean +@ServiceActivator(inputChannel = "inboundStaging") +public AggregatorFactoryBean partitioningMessageHandler() throws Exception { + AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); + aggregatorFactoryBean.setProcessorBean(partitionHandler()); + aggregatorFactoryBean.setOutputChannel(outboundReplies()); + ... + return aggregatorFactoryBean; +} + +@Bean +public QueueChannel outboundReplies() { + return new QueueChannel(); +} +---- + You must also ensure that the partition `handler` attribute maps to the `partitionHandler` bean, as shown in the following example: -[source, xml] +.XML Configuration +[source, xml, role="xmlContent"] ---- @@ -807,3 +1161,15 @@ You must also ensure that the partition `handler` attribute maps to the `partiti ---- + +.Java Configuration +[source, java, role="javaContent"] +---- + public Job personJob() { + return jobBuilderFactory.get("personJob") + .start(stepBuilderFactory.get("step1.master") + .partitioner(partitionHandler()) + .build()) + .build(); + } +----