From 450de51d0c003240056befc5c538217eb3ca6719 Mon Sep 17 00:00:00 2001 From: Ali Shahbour Date: Mon, 6 Mar 2017 17:06:51 +0200 Subject: [PATCH] Add Ordered interface to all event listeneres changes requested by cppwfs Fix Constructors duplication Add @author Some cleanup on chunk listener Updated chunk listener to send message before and after chunk. Fixed Links for JavaDocs to prevent failure on javadoc build. Updated tests to support new chunk messages. --- .../src/main/asciidoc/stream.adoc | 15 ++- .../listener/BatchExecutionEventTests.java | 2 +- .../listener/BatchEventAutoConfiguration.java | 29 +++-- .../listener/EventEmittingChunkListener.java | 72 +++++++++++ .../EventEmittingItemProcessListener.java | 15 ++- .../EventEmittingItemReadListener.java | 15 ++- .../EventEmittingItemWriteListener.java | 15 ++- .../EventEmittingJobExecutionListener.java | 15 ++- .../listener/EventEmittingSkipListener.java | 15 ++- .../EventEmittingStepExecutionListener.java | 15 ++- .../listener/support/TaskEventProperties.java | 118 ++++++++++++++++++ .../batch/listener/EventListenerTests.java | 53 +++++++- .../listener/JobExecutionEventTests.java | 25 ++++ 13 files changed, 382 insertions(+), 22 deletions(-) create mode 100644 spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java create mode 100644 spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java 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" : "";