diff --git a/spring-cloud-task-docs/src/main/asciidoc/stream.adoc b/spring-cloud-task-docs/src/main/asciidoc/stream.adoc index f386b506..13577e4a 100644 --- a/spring-cloud-task-docs/src/main/asciidoc/stream.adoc +++ b/spring-cloud-task-docs/src/main/asciidoc/stream.adoc @@ -189,7 +189,7 @@ spring.cloud.task.batch.events.enabled=false ``` To disable a specific batch event use the following configuration: -`spring.cloud.task.events..enabled=false`: +`spring.cloud.task.batch.events..enabled=false`: ``` spring.cloud.task.batch.events.job-execution.enabled=false @@ -200,3 +200,16 @@ spring.cloud.task.batch.events.item-process.enabled=false spring.cloud.task.batch.events.item-write.enabled=false spring.cloud.task.batch.events.skip.enabled=false ``` + +=== Emit Order for Batch Events +By default batch events have `Ordered.LOWEST_PRECEDENCE` , to change this value ( for example to 5 ) use the following configuration: + +``` +spring.cloud.task.batch.events.job-execution-order=5 +spring.cloud.task.batch.events.step-execution-order=5 +spring.cloud.task.batch.events.chunk-order=5 +spring.cloud.task.batch.events.item-read-order=5 +spring.cloud.task.batch.events.item-process-order=5 +spring.cloud.task.batch.events.item-write-order=5 +spring.cloud.task.batch.events.skip-order=5 +``` \ No newline at end of file diff --git a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java b/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java index 5ae8b8f7..e6facd04 100644 --- a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java +++ b/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java @@ -201,7 +201,7 @@ public class BatchExecutionEventTests { public static class ChunkEventsListenerBinding { @StreamListener(Sink.INPUT) - public void receive(ChunkContext chunkContext) { + public void receive(String message) { chunkEventsLatch.countDown(); } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java index fceae03d..a7fe8f93 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java @@ -28,9 +28,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.task.batch.listener.support.TaskBatchEventListenerBeanPostProcessor; +import org.springframework.cloud.task.batch.listener.support.TaskEventProperties; import org.springframework.cloud.task.listener.TaskLifecycleListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -53,6 +56,7 @@ import org.springframework.messaging.MessageChannel; * * @author Michael Minella * @author Glenn Renfro + * @author Ali Shahbour */ @Configuration @ConditionalOnClass(Job.class) @@ -77,59 +81,58 @@ public class BatchEventAutoConfiguration { @Configuration @ConditionalOnClass(EnableBinding.class) @EnableBinding(BatchEventsChannels.class) + @EnableConfigurationProperties(TaskEventProperties.class) @ConditionalOnMissingBean(name = JOB_EXECUTION_EVENTS_LISTENER) public static class JobExecutionListenerConfiguration { @Autowired private BatchEventsChannels listenerChannels; + @Autowired + private TaskEventProperties taskEventProperties; + @Bean @Lazy @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.job-execution", name = "enabled", havingValue = "true", matchIfMissing = true) public JobExecutionListener jobExecutionEventsListener() { - return new EventEmittingJobExecutionListener(listenerChannels.jobExecutionEvents()); + return new EventEmittingJobExecutionListener(listenerChannels.jobExecutionEvents(),taskEventProperties.getJobExecutionOrder()); } @Bean @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.step-execution", name = "enabled", havingValue = "true", matchIfMissing = true) public StepExecutionListener stepExecutionEventsListener() { - return new EventEmittingStepExecutionListener(listenerChannels.stepExecutionEvents()); + return new EventEmittingStepExecutionListener(listenerChannels.stepExecutionEvents(),taskEventProperties.getStepExecutionOrder()); } @Bean @Lazy @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.chunk", name = "enabled", havingValue = "true", matchIfMissing = true) - public GatewayProxyFactoryBean chunkEventsListener() { - GatewayProxyFactoryBean factoryBean = - new GatewayProxyFactoryBean(ChunkListener.class); - - factoryBean.setDefaultRequestChannel(listenerChannels.chunkEvents()); - - return factoryBean; + public EventEmittingChunkListener chunkEventsListener() { + return new EventEmittingChunkListener(listenerChannels.chunkEvents(),taskEventProperties.getChunkOrder()); } @Bean @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-read", name = "enabled", havingValue = "true", matchIfMissing = true) public ItemReadListener itemReadEventsListener() { - return new EventEmittingItemReadListener(listenerChannels.itemReadEvents()); + return new EventEmittingItemReadListener(listenerChannels.itemReadEvents(),taskEventProperties.getItemReadOrder()); } @Bean @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-write", name = "enabled", havingValue = "true", matchIfMissing = true) public ItemWriteListener itemWriteEventsListener() { - return new EventEmittingItemWriteListener(listenerChannels.itemWriteEvents()); + return new EventEmittingItemWriteListener(listenerChannels.itemWriteEvents(),taskEventProperties.getItemWriteOrder()); } @Bean @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-process", name = "enabled", havingValue = "true", matchIfMissing = true) public ItemProcessListener itemProcessEventsListener() { - return new EventEmittingItemProcessListener(listenerChannels.itemProcessEvents()); + return new EventEmittingItemProcessListener(listenerChannels.itemProcessEvents(),taskEventProperties.getItemProcessOrder()); } @Bean @ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.skip", name = "enabled", havingValue = "true", matchIfMissing = true) public SkipListener skipEventsListener() { - return new EventEmittingSkipListener(listenerChannels.skipEvents()); + return new EventEmittingSkipListener(listenerChannels.skipEvents(),taskEventProperties.getItemProcessOrder()); } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java new file mode 100644 index 00000000..34df52bb --- /dev/null +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java @@ -0,0 +1,72 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.cloud.task.batch.listener; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * Provides informational messages around the {@link org.springframework.batch.core.step.item.Chunk} of a batch job. + * + * The {@link ChunkListener#beforeChunk(ChunkContext)} and + * {@link ChunkListener#afterChunk(ChunkContext)} are both no-ops in this implementation. + * {@link ChunkListener#afterChunkError(ChunkContext)}. + * + * @author Ali Shahbour + */ +public class EventEmittingChunkListener implements ChunkListener, Ordered { + + private static final Log logger = LogFactory.getLog(EventEmittingChunkListener.class); + + private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; + + public EventEmittingChunkListener(MessageChannel output) { + Assert.notNull(output, "An output channel is required"); + this.messagePublisher = new MessagePublisher(output); + } + + public EventEmittingChunkListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + + @Override + public void beforeChunk(ChunkContext context) { + messagePublisher.publish("Before Chunk Processing"); + } + + @Override + public void afterChunk(ChunkContext context) { + messagePublisher.publish("After Chunk Processing"); + } + + @Override + public void afterChunkError(ChunkContext context) { + + } + + @Override + public int getOrder() { + return this.order; + } +} diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java index de5c737e..53960b88 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java @@ -18,6 +18,7 @@ package org.springframework.cloud.task.batch.listener; import org.springframework.batch.core.ItemProcessListener; import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -33,16 +34,23 @@ import org.springframework.util.Assert; * * @author Michael Minella * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingItemProcessListener implements ItemProcessListener { +public class EventEmittingItemProcessListener implements ItemProcessListener, Ordered { private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingItemProcessListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher<>(output); } + public EventEmittingItemProcessListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void beforeProcess(Object item) { } @@ -64,4 +72,9 @@ public class EventEmittingItemProcessListener implements ItemProcessListener { public void onProcessError(Object item, Exception e) { messagePublisher.publishWithThrowableHeader("Exception while item was being processed", e.getMessage()); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java index afb351e8..783bb928 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java @@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ItemReadListener; import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -34,18 +35,25 @@ import org.springframework.util.Assert; * via the {@link BatchJobHeaders.BATCH_EXCEPTION} message header. * * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingItemReadListener implements ItemReadListener { +public class EventEmittingItemReadListener implements ItemReadListener, Ordered { private static final Log logger = LogFactory.getLog(EventEmittingItemReadListener.class); private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingItemReadListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher(output); } + public EventEmittingItemReadListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void beforeRead() { @@ -64,4 +72,9 @@ public class EventEmittingItemReadListener implements ItemReadListener { messagePublisher.publishWithThrowableHeader("Exception while item was being read", ex.getMessage()); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java index d072ba68..21b7b36e 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java @@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ItemWriteListener; import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -35,18 +36,25 @@ import org.springframework.util.Assert; * the exception's message via the {@link BatchJobHeaders.BATCH_EXCEPTION} message header. * * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingItemWriteListener implements ItemWriteListener{ +public class EventEmittingItemWriteListener implements ItemWriteListener, Ordered { private static final Log logger = LogFactory.getLog(EventEmittingItemWriteListener.class); private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingItemWriteListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher<>(output); } + public EventEmittingItemWriteListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void beforeWrite(List items) { this.messagePublisher.publish(items.size() + " items to be written."); @@ -68,4 +76,9 @@ public class EventEmittingItemWriteListener implements ItemWriteListener{ String payload = "Exception while " + items.size() + " items are attempted to be written."; this.messagePublisher.publishWithThrowableHeader(payload, exception.getMessage()); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java index ae1c19bb..8b52830c 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java @@ -19,6 +19,7 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobExecutionListener; import org.springframework.cloud.task.batch.listener.support.JobExecutionEvent; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -27,16 +28,23 @@ import org.springframework.util.Assert; * * @author Michael Minella * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingJobExecutionListener implements JobExecutionListener { +public class EventEmittingJobExecutionListener implements JobExecutionListener, Ordered { private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingJobExecutionListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher<>(output); } + public EventEmittingJobExecutionListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void beforeJob(JobExecution jobExecution) { this.messagePublisher.publish(new JobExecutionEvent(jobExecution)); @@ -46,4 +54,9 @@ public class EventEmittingJobExecutionListener implements JobExecutionListener { public void afterJob(JobExecution jobExecution) { this.messagePublisher.publish(new JobExecutionEvent(jobExecution)); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java index a8921186..084578c8 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java @@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.SkipListener; import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -35,18 +36,25 @@ import org.springframework.util.Assert; * of the item that caused the error. * * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingSkipListener implements SkipListener { +public class EventEmittingSkipListener implements SkipListener, Ordered { private static final Log logger = LogFactory.getLog(EventEmittingSkipListener.class); private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingSkipListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher<>(output); } + public EventEmittingSkipListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void onSkipInRead(Throwable t) { if (logger.isDebugEnabled()) { @@ -70,4 +78,9 @@ public class EventEmittingSkipListener implements SkipListener { } messagePublisher.publishWithThrowableHeader(item, t.getMessage()); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java index 42854e06..78a351a9 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java @@ -20,6 +20,7 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent; import org.springframework.cloud.task.batch.listener.support.MessagePublisher; +import org.springframework.core.Ordered; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -30,16 +31,23 @@ import org.springframework.util.Assert; * * @author Michael Minella * @author Glenn Renfro + * @author Ali Shahbour */ -public class EventEmittingStepExecutionListener implements StepExecutionListener { +public class EventEmittingStepExecutionListener implements StepExecutionListener, Ordered { private MessagePublisher messagePublisher; + private int order = Ordered.LOWEST_PRECEDENCE; public EventEmittingStepExecutionListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); this.messagePublisher = new MessagePublisher<>(output); } + public EventEmittingStepExecutionListener(MessageChannel output, int order) { + this(output); + this.order = order; + } + @Override public void beforeStep(StepExecution stepExecution) { this.messagePublisher.publish(new StepExecutionEvent(stepExecution)); @@ -51,4 +59,9 @@ public class EventEmittingStepExecutionListener implements StepExecutionListener return stepExecution.getExitStatus(); } + + @Override + public int getOrder() { + return this.order; + } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java new file mode 100644 index 00000000..7a96fe3e --- /dev/null +++ b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java @@ -0,0 +1,118 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.task.batch.listener.support; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.core.Ordered; + +/** + * @author Ali Shahbour + */ +@ConfigurationProperties(prefix = "spring.cloud.task.batch.events") +public class TaskEventProperties { + + /** + * Properties for jobExecution listener order + */ + private int jobExecutionOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for stepExecution listener order + */ + private int stepExecutionOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for itemRead listener order + */ + private int itemReadOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for itemProcess listener order + */ + private int itemProcessOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for itemWrite listener order + */ + private int itemWriteOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for chunk listener order + */ + private int chunkOrder = Ordered.LOWEST_PRECEDENCE; + + /** + * Properties for skip listener order + */ + private int skipOrder = Ordered.LOWEST_PRECEDENCE; + + public int getJobExecutionOrder() { + return jobExecutionOrder; + } + + public void setJobExecutionOrder(int jobExecutionOrder) { + this.jobExecutionOrder = jobExecutionOrder; + } + + public int getStepExecutionOrder() { + return stepExecutionOrder; + } + + public void setStepExecutionOrder(int stepExecutionOrder) { + this.stepExecutionOrder = stepExecutionOrder; + } + + public int getItemReadOrder() { + return itemReadOrder; + } + + public void setItemReadOrder(int itemReadOrder) { + this.itemReadOrder = itemReadOrder; + } + + public int getItemProcessOrder() { + return itemProcessOrder; + } + + public void setItemProcessOrder(int itemProcessOrder) { + this.itemProcessOrder = itemProcessOrder; + } + + public int getItemWriteOrder() { + return itemWriteOrder; + } + + public void setItemWriteOrder(int itemWriteOrder) { + this.itemWriteOrder = itemWriteOrder; + } + + public int getChunkOrder() { + return chunkOrder; + } + + public void setChunkOrder(int chunkOrder) { + this.chunkOrder = chunkOrder; + } + + public int getSkipOrder() { + return skipOrder; + } + + public void setSkipOrder(int skipOrder) { + this.skipOrder = skipOrder; + } +} diff --git a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventListenerTests.java b/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventListenerTests.java index 25f274a5..7bd4581f 100644 --- a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventListenerTests.java +++ b/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventListenerTests.java @@ -27,8 +27,11 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.batch.core.scope.context.StepContext; import org.springframework.cloud.task.batch.listener.support.JobExecutionEvent; import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent; +import org.springframework.core.Ordered; import org.springframework.integration.channel.QueueChannel; import org.springframework.messaging.Message; @@ -36,6 +39,7 @@ import static org.junit.Assert.assertEquals; /** * @author Glenn Renfro + * @author Ali Shahbour */ public class EventListenerTests { @@ -47,7 +51,7 @@ public class EventListenerTests { private EventEmittingItemWriteListener eventEmittingItemWriteListener; private EventEmittingJobExecutionListener eventEmittingJobExecutionListener; private EventEmittingStepExecutionListener eventEmittingStepExecutionListener; - + private EventEmittingChunkListener eventEmittingChunkListener; @Before public void beforeTests() { @@ -58,8 +62,21 @@ public class EventListenerTests { eventEmittingItemWriteListener = new EventEmittingItemWriteListener(queueChannel); eventEmittingJobExecutionListener = new EventEmittingJobExecutionListener(queueChannel); eventEmittingStepExecutionListener = new EventEmittingStepExecutionListener(queueChannel); + eventEmittingChunkListener = new EventEmittingChunkListener(queueChannel,0); } + @Test + public void testEventListenerOrderProperty() { + assertEquals(eventEmittingSkipListener.getOrder(),Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingItemProcessListener.getOrder(), Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingItemReadListener.getOrder(), Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingItemWriteListener.getOrder(), Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingJobExecutionListener.getOrder(), Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingStepExecutionListener.getOrder(), Ordered.LOWEST_PRECEDENCE); + assertEquals(eventEmittingChunkListener.getOrder(),0); + } + + @Test public void testItemProcessListenerOnProcessorError() { RuntimeException exeption = new RuntimeException("Test Exception"); @@ -219,6 +236,33 @@ public class EventListenerTests { stepExecutionEvent.getStepName()); } + @Test + public void EventEmittingChunkExecutionListenerBeforeChunk() { + final String CHUNK_MESSAGE = "Before Chunk Processing"; + ChunkContext chunkContext = getChunkContext(); + eventEmittingChunkListener.beforeChunk(chunkContext); + assertEquals(1,queueChannel.getQueueSize()); + Message msg = queueChannel.receive(); + assertEquals(CHUNK_MESSAGE,msg.getPayload()); + } + + @Test + public void EventEmittingChunkExecutionListenerAfterChunk() { + final String CHUNK_MESSAGE = "After Chunk Processing"; + ChunkContext chunkContext = getChunkContext(); + eventEmittingChunkListener.afterChunk(chunkContext); + assertEquals(1,queueChannel.getQueueSize()); + Message msg = queueChannel.receive(); + assertEquals(CHUNK_MESSAGE,msg.getPayload()); + } + + @Test + public void EventEmittingChunkExecutionListenerAfterChunkError() { + ChunkContext chunkContext = getChunkContext(); + eventEmittingChunkListener.afterChunkError(chunkContext); + assertEquals(0,queueChannel.getQueueSize()); + } + private JobExecution getJobExecution() { final String JOB_NAME = UUID.randomUUID().toString(); JobInstance jobInstance = new JobInstance(1L, JOB_NAME); @@ -232,5 +276,12 @@ public class EventListenerTests { testList.add("foo"); return testList; } + private ChunkContext getChunkContext() { + JobExecution jobExecution = getJobExecution(); + StepExecution stepExecution = new StepExecution("STEP1",jobExecution); + StepContext stepContext = new StepContext(stepExecution); + ChunkContext chunkContext = new ChunkContext(stepContext); + return chunkContext; + } } diff --git a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/JobExecutionEventTests.java b/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/JobExecutionEventTests.java index dabb5844..0a3b92e7 100644 --- a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/JobExecutionEventTests.java +++ b/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/JobExecutionEventTests.java @@ -43,6 +43,7 @@ import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent; import org.springframework.cloud.task.configuration.EnableTask; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Configuration; +import org.springframework.core.Ordered; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -51,6 +52,7 @@ import static org.junit.Assert.assertTrue; /** * @author Glenn Renfro. + * @author Ali Shahbour */ public class JobExecutionEventTests { @@ -270,6 +272,29 @@ public class JobExecutionEventTests { assertEquals(BatchStatus.FAILED, jobExecutionEvent.getStatus()); } + @Test + public void testOrderConfiguration() { + ConfigurableApplicationContext applicationContext = + SpringApplication.run(new Object[]{BatchEventAutoConfiguration.JobExecutionListenerConfiguration.class, + EventJobExecutionConfiguration.class, + PropertyPlaceholderAutoConfiguration.class, + TestSupportBinderAutoConfiguration.class}, + new String[]{"--spring.cloud.task.closecontext_enable=false", + "--spring.main.web-environment=false", + "--spring.cloud.task.batch.events.chunk-order=5", + "--spring.cloud.task.batch.events.item-process-order=5", + "--spring.cloud.task.batch.events.item-read-order=5", + "--spring.cloud.task.batch.events.item-write-order=5", + "--spring.cloud.task.batch.events.job-execution-order=5", + "--spring.cloud.task.batch.events.skip-order=5", + "--spring.cloud.task.batch.events.step-execution-order=5" + }); + for (String beanName : LISTENER_BEAN_NAMES) { + Ordered ordered = (Ordered)applicationContext.getBean(beanName); + assertEquals("Expected order value of 5 for " + beanName,ordered.getOrder(),5); + } + } + public void testDisabledConfiguration(String property, String disabledListener) { boolean exceptionThrown = false; String disabledPropertyArg = (property != null) ? "--" + property + "=false" : "";