From 43943dcae5b9c4333be9d05192934771158eaeb8 Mon Sep 17 00:00:00 2001 From: Glenn Renfro Date: Thu, 23 Jun 2016 15:57:12 -0400 Subject: [PATCH] Fixes BeanNotFoundException when disabling BatchEventListeners. Resolves #153 --- .../src/main/asciidoc/stream.adoc | 23 ++++- ...skBatchEventListenerBeanPostProcessor.java | 90 ++++++++++++++----- 2 files changed, 91 insertions(+), 22 deletions(-) diff --git a/spring-cloud-task-docs/src/main/asciidoc/stream.adoc b/spring-cloud-task-docs/src/main/asciidoc/stream.adoc index dfa5a514..ac2c37fc 100644 --- a/spring-cloud-task-docs/src/main/asciidoc/stream.adoc +++ b/spring-cloud-task-docs/src/main/asciidoc/stream.adoc @@ -141,8 +141,29 @@ spring.cloud.stream.bindings.input.destination=job-execution-events NOTE: A binder implementation is also required to be on the classpath. -To disable the listener functionality, use the following configuration: +=== Sending Batch Events to different channels + +One of the options that Spring Cloud Task offers for batch events is the ability to alter the channel to which a +specific listener can emit its messages. To do this use the following configuration: +`spring.cloud.stream.bindings..destination=`. +For example: If StepExecutionListener needs to emit its messages to another channel `my-step-execution-events` +instead of the default `step-execution-events` the following configuration can be added: + +``` +spring.cloud.stream.bindings.step-execution-events.destination=my-step-execution-events` +``` + +=== Disabling Batch Events +To disable the all batch event listener functionality, use the following configuration: ``` spring.cloud.task.batch.events.enabled=false ``` + +To disable a specific batch event use the following configuration: +`spring.cloud.task.events..enabled=false`. For example if the ItemWriteListener should not emit +events add the following configuration: + +``` +spring.cloud.task.events.item.write.events.listener.enabled=false +``` diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java index 77b5fd0a..4ab24d9f 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java @@ -32,6 +32,7 @@ import org.springframework.batch.core.step.item.SimpleChunkProvider; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.cloud.task.batch.listener.BatchEventAutoConfiguration; @@ -66,27 +67,14 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof AbstractJob) { - JobExecutionListener jobExecutionEventsListener = - (JobExecutionListener) this.applicationContext.getBean( - BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER); - - AbstractJob job = (AbstractJob) bean; - job.registerJobExecutionListener( - jobExecutionEventsListener); - } + registerJobExecutionEventListener(bean); if (bean instanceof AbstractStep) { - StepExecutionListener stepExecutionListener = - (StepExecutionListener) this.applicationContext.getBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER); - - AbstractStep step = (AbstractStep) bean; - step.registerStepExecutionListener(stepExecutionListener); - + registerStepExecutionEventListener(bean); if (bean instanceof TaskletStep) { TaskletStep taskletStep = (TaskletStep) bean; - taskletStep.registerChunkListener((ChunkListener) this.applicationContext.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER)); Tasklet tasklet = taskletStep.getTasklet(); + registerChunkEventsListener(bean); if (tasklet instanceof ChunkOrientedTasklet) { Field chunkProviderField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProvider"); @@ -95,11 +83,11 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso Field chunkProcessorField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProcessor"); ReflectionUtils.makeAccessible(chunkProcessorField); SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils.getField(chunkProcessorField, tasklet); - chunkProvider.registerListener((ItemReadListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER)); - chunkProvider.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)); - chunkProcessor.registerListener((ItemProcessListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER)); - chunkProcessor.registerListener((ItemWriteListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER)); - chunkProcessor.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)); + registerItemReadEvents(chunkProvider); + registerSkipEvents(chunkProvider); + registerItemProcessEvents(chunkProcessor); + registerItemWriteEvents(chunkProcessor); + registerSkipEvents(chunkProcessor); } } } @@ -111,4 +99,64 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { return bean; } + + private void registerItemProcessEvents(SimpleChunkProcessor chunkProcessor) { + if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER)) { + chunkProcessor.registerListener((ItemProcessListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER)); + } + } + + private void registerItemReadEvents(SimpleChunkProvider chunkProvider) { + if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER)) { + chunkProvider.registerListener((ItemReadListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER)); + } + } + + private void registerItemWriteEvents(SimpleChunkProcessor chunkProcessor) { + if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER)) { + chunkProcessor.registerListener((ItemWriteListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER)); + } + } + + private void registerSkipEvents(SimpleChunkProvider chunkProvider) { + if (this.applicationContext.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) { + chunkProvider.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)); + } + } + + private void registerSkipEvents(SimpleChunkProcessor chunkProcessor) { + if(this.applicationContext.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) { + chunkProcessor.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)); + } + } + + private void registerChunkEventsListener(Object bean) { + if(this.applicationContext.containsBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER)) + { + ((TaskletStep)bean).registerChunkListener((ChunkListener) + this.applicationContext.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER)); + } + } + + private void registerJobExecutionEventListener(Object bean) { + if (bean instanceof AbstractJob && + this.applicationContext.containsBean(BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER)) { + JobExecutionListener jobExecutionEventsListener = + (JobExecutionListener) this.applicationContext.getBean( + BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER); + + AbstractJob job = (AbstractJob) bean; + job.registerJobExecutionListener( + jobExecutionEventsListener); + } + } + + private void registerStepExecutionEventListener(Object bean) { + if (this.applicationContext.containsBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER)) { + StepExecutionListener stepExecutionListener = + (StepExecutionListener) this.applicationContext.getBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER); + AbstractStep step = (AbstractStep) bean; + step.registerStepExecutionListener(stepExecutionListener); + } + } }