Base Task Sink Case with Tests works

In process of migrating batch-events to streambridge

BatchEvents Migrated

Integration tests for TaskSink now work

Batch Event Integration Tests updated

Added tests to batch events

Event integration tests added

Updated with the last bit of tests

baseline polishing

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java

@@ -32,8 +32,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 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.stream.function.StreamBridge;
+import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
 import org.springframework.cloud.task.batch.listener.support.TaskBatchEventListenerBeanPostProcessor;
 import org.springframework.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.cloud.task.configuration.SimpleTaskAutoConfiguration;
@@ -41,7 +41,6 @@ import org.springframework.cloud.task.listener.TaskLifecycleListener;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Lazy;
-import org.springframework.messaging.MessageChannel;
 
 /**
  * Autoconfigures Spring Batch listeners designed to emit events on the following
@@ -112,82 +111,16 @@ public class BatchEventAutoConfiguration {
 		return new TaskBatchEventListenerBeanPostProcessor();
 	}
 
-	/**
-	 * Name of Batch Events channels.
-	 */
-	public interface BatchEventsChannels {
-
-		/**
-		 * Name of the job execution events channel.
-		 */
-		String JOB_EXECUTION_EVENTS = "job-execution-events";
-
-		/**
-		 * Name of the step execution events channel.
-		 */
-		String STEP_EXECUTION_EVENTS = "step-execution-events";
-
-		/**
-		 * Name of the chunk execution events channel.
-		 */
-		String CHUNK_EXECUTION_EVENTS = "chunk-events";
-
-		/**
-		 * Name of the item read events channel.
-		 */
-		String ITEM_READ_EVENTS = "item-read-events";
-
-		/**
-		 * Name of the item process events channel.
-		 */
-		String ITEM_PROCESS_EVENTS = "item-process-events";
-
-		/**
-		 * Name of the item write events channel.
-		 */
-		String ITEM_WRITE_EVENTS = "item-write-events";
-
-		/**
-		 * Name of the skip events channel.
-		 */
-		String SKIP_EVENTS = "skip-events";
-
-		@Output(JOB_EXECUTION_EVENTS)
-		MessageChannel jobExecutionEvents();
-
-		@Output(STEP_EXECUTION_EVENTS)
-		MessageChannel stepExecutionEvents();
-
-		@Output(CHUNK_EXECUTION_EVENTS)
-		MessageChannel chunkEvents();
-
-		@Output(ITEM_READ_EVENTS)
-		MessageChannel itemReadEvents();
-
-		@Output(ITEM_WRITE_EVENTS)
-		MessageChannel itemWriteEvents();
-
-		@Output(ITEM_PROCESS_EVENTS)
-		MessageChannel itemProcessEvents();
-
-		@Output(SKIP_EVENTS)
-		MessageChannel skipEvents();
-
-	}
-
 	/**
 	 * Configuration for Job Execution Listener.
 	 */
 	@Configuration(proxyBeanMethods = false)
-	@ConditionalOnClass(EnableBinding.class)
-	@EnableBinding(BatchEventsChannels.class)
+	@ConditionalOnClass(StreamBridge.class)
 	@EnableConfigurationProperties(TaskEventProperties.class)
 	@ConditionalOnMissingBean(name = JOB_EXECUTION_EVENTS_LISTENER)
 	@ConditionalOnExpression("T(org.springframework.util.StringUtils).isEmpty('${spring.batch.job.jobName:}')")
 	public static class JobExecutionListenerConfiguration {
 
-		@Autowired
-		private BatchEventsChannels listenerChannels;
 
 		@Autowired
 		private TaskEventProperties taskEventProperties;
@@ -198,10 +131,10 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.job-execution",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public JobExecutionListener jobExecutionEventsListener() {
+		public JobExecutionListener jobExecutionEventsListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
 			return new EventEmittingJobExecutionListener(
-					this.listenerChannels.jobExecutionEvents(),
-					this.taskEventProperties.getJobExecutionOrder());
+					messagePublisher,
+					this.taskEventProperties.getJobExecutionOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -209,10 +142,10 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.step-execution",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public StepExecutionListener stepExecutionEventsListener() {
+		public StepExecutionListener stepExecutionEventsListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
 			return new EventEmittingStepExecutionListener(
-					this.listenerChannels.stepExecutionEvents(),
-					this.taskEventProperties.getStepExecutionOrder());
+					messagePublisher,
+					this.taskEventProperties.getStepExecutionOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -221,9 +154,9 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.chunk",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public EventEmittingChunkListener chunkEventsListener() {
-			return new EventEmittingChunkListener(this.listenerChannels.chunkEvents(),
-					this.taskEventProperties.getChunkOrder());
+		public EventEmittingChunkListener chunkEventsListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+			return new EventEmittingChunkListener(messagePublisher,
+					this.taskEventProperties.getChunkOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -231,10 +164,10 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-read",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public ItemReadListener itemReadEventsListener() {
+		public ItemReadListener itemReadEventsListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
 			return new EventEmittingItemReadListener(
-					this.listenerChannels.itemReadEvents(),
-					this.taskEventProperties.getItemReadOrder());
+					messagePublisher,
+					this.taskEventProperties.getItemReadOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -242,10 +175,11 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-write",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public ItemWriteListener itemWriteEventsListener() {
+		public ItemWriteListener itemWriteEventsListener(MessagePublisher messagePublisher,
+			TaskEventProperties properties) {
 			return new EventEmittingItemWriteListener(
-					this.listenerChannels.itemWriteEvents(),
-					this.taskEventProperties.getItemWriteOrder());
+					messagePublisher,
+					this.taskEventProperties.getItemWriteOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -253,10 +187,11 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-process",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public ItemProcessListener itemProcessEventsListener() {
+		public ItemProcessListener itemProcessEventsListener(MessagePublisher messagePublisher,
+			TaskEventProperties properties) {
 			return new EventEmittingItemProcessListener(
-					this.listenerChannels.itemProcessEvents(),
-					this.taskEventProperties.getItemProcessOrder());
+					messagePublisher,
+					this.taskEventProperties.getItemProcessOrder(), properties);
 		}
 
 		// @checkstyle:off
@@ -264,9 +199,15 @@ public class BatchEventAutoConfiguration {
 		@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.skip",
 				name = "enabled", havingValue = "true", matchIfMissing = true)
 		// @checkstyle:on
-		public SkipListener skipEventsListener() {
-			return new EventEmittingSkipListener(this.listenerChannels.skipEvents(),
-					this.taskEventProperties.getItemProcessOrder());
+		public SkipListener skipEventsListener(MessagePublisher messagePublisher,
+			TaskEventProperties properties) {
+			return new EventEmittingSkipListener(messagePublisher,
+					this.taskEventProperties.getItemProcessOrder(), properties);
+		}
+
+		@Bean
+		public MessagePublisher messagePublisher(StreamBridge streamBridge) {
+			return new MessagePublisher(streamBridge);
 		}
 
 	}

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -16,14 +16,11 @@
 
 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.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -38,30 +35,34 @@ import org.springframework.util.Assert;
  */
 public class EventEmittingChunkListener implements ChunkListener, Ordered {
 
-	private static final Log logger = LogFactory.getLog(EventEmittingChunkListener.class);
-
-	private MessagePublisher<String> messagePublisher;
-
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingChunkListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher(output);
+	private MessagePublisher messagePublisher;
+
+	private TaskEventProperties properties;
+
+	public EventEmittingChunkListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingChunkListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingChunkListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
 	@Override
 	public void beforeChunk(ChunkContext context) {
-		this.messagePublisher.publish("Before Chunk Processing");
+		this.messagePublisher.publish(this.properties.getChunkEventBindingName(),
+			"Before Chunk Processing");
 	}
 
 	@Override
 	public void afterChunk(ChunkContext context) {
-		this.messagePublisher.publish("After Chunk Processing");
+		this.messagePublisher.publish(this.properties.getChunkEventBindingName(),
+			"After Chunk Processing");
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -16,12 +16,15 @@
 
 package org.springframework.cloud.task.batch.listener;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.batch.core.ItemProcessListener;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
+import org.springframework.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -40,17 +43,24 @@ import org.springframework.util.Assert;
  */
 public class EventEmittingItemProcessListener implements ItemProcessListener, Ordered {
 
-	private MessagePublisher<String> messagePublisher;
+	private static final Log logger = LogFactory
+		.getLog(EventEmittingItemProcessListener.class);
+
+	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);
+	private TaskEventProperties properties;
+
+	public EventEmittingItemProcessListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingItemProcessListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingItemProcessListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
@@ -61,20 +71,24 @@ public class EventEmittingItemProcessListener implements ItemProcessListener, Or
 	@Override
 	public void afterProcess(Object item, Object result) {
 		if (result == null) {
-			this.messagePublisher.publish("1 item was filtered");
+			this.messagePublisher.publish(this.properties.getItemProcessEventBindingName(), "1 item was filtered");
 		}
 		else if (item.equals(result)) {
-			this.messagePublisher.publish("item equaled result after processing");
+			this.messagePublisher.publish(this.properties.getItemProcessEventBindingName(), "item equaled result after processing");
 		}
 		else {
-			this.messagePublisher.publish("item did not equal result after processing");
+			this.messagePublisher.publish(this.properties.getItemProcessEventBindingName(), "item did not equal result after processing");
 		}
 	}
 
 	@Override
 	public void onProcessError(Object item, Exception e) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onProcessError: " + e.getMessage(), e);
+		}
 		this.messagePublisher.publishWithThrowableHeader(
-				"Exception while item was being processed", e.getMessage());
+			this.properties.getItemProcessEventBindingName(),
+			"Exception while item was being processed", e.getMessage());
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -23,8 +23,8 @@ import org.springframework.batch.core.ItemReadListener;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
+import org.springframework.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -43,17 +43,22 @@ public class EventEmittingItemReadListener implements ItemReadListener, Ordered
 	private static final Log logger = LogFactory
 			.getLog(EventEmittingItemReadListener.class);
 
-	private MessagePublisher<String> messagePublisher;
-
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingItemReadListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher(output);
+	private final MessagePublisher<String> messagePublisher;
+
+	private TaskEventProperties properties;
+
+	public EventEmittingItemReadListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+		this.properties = properties;
+		this.messagePublisher = messagePublisher;
 	}
 
-	public EventEmittingItemReadListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingItemReadListener(MessagePublisher messagePublisher,
+		int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
@@ -73,7 +78,7 @@ public class EventEmittingItemReadListener implements ItemReadListener, Ordered
 			logger.debug("Executing onReadError: " + ex.getMessage(), ex);
 		}
 
-		this.messagePublisher.publishWithThrowableHeader(
+		this.messagePublisher.publishWithThrowableHeader(this.properties.getItemReadEventBindingName(),
 				"Exception while item was being read", ex.getMessage());
 	}
 

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -24,8 +24,8 @@ 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.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -44,23 +44,29 @@ public class EventEmittingItemWriteListener implements ItemWriteListener, Ordere
 	private static final Log logger = LogFactory
 			.getLog(EventEmittingItemWriteListener.class);
 
-	private MessagePublisher<String> messagePublisher;
-
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingItemWriteListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher<>(output);
+	private final MessagePublisher<String> messagePublisher;
+
+	private TaskEventProperties properties;
+
+	public EventEmittingItemWriteListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingItemWriteListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingItemWriteListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
 	@Override
 	public void beforeWrite(List items) {
-		this.messagePublisher.publish(items.size() + " items to be written.");
+		this.messagePublisher.publish(this.properties.getItemWriteEventBindingName(),
+			items.size() + " items to be written.");
 	}
 
 	@Override
@@ -68,7 +74,8 @@ public class EventEmittingItemWriteListener implements ItemWriteListener, Ordere
 		if (logger.isDebugEnabled()) {
 			logger.debug("Executing afterWrite: " + items);
 		}
-		this.messagePublisher.publish(items.size() + " items have been written.");
+		this.messagePublisher.publish(this.properties.getItemWriteEventBindingName(),
+			items.size() + " items have been written.");
 	}
 
 	@Override
@@ -78,7 +85,8 @@ public class EventEmittingItemWriteListener implements ItemWriteListener, Ordere
 		}
 		String payload = "Exception while " + items.size()
 				+ " items are attempted to be written.";
-		this.messagePublisher.publishWithThrowableHeader(payload, exception.getMessage());
+		this.messagePublisher.publishWithThrowableHeader(
+			this.properties.getItemWriteEventBindingName(), payload, exception.getMessage());
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -20,8 +20,8 @@ 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.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -33,28 +33,33 @@ import org.springframework.util.Assert;
  */
 public class EventEmittingJobExecutionListener implements JobExecutionListener, Ordered {
 
-	private MessagePublisher<JobExecutionEvent> messagePublisher;
-
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingJobExecutionListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher<>(output);
+	private final MessagePublisher<JobExecutionEvent> messagePublisher;
+
+	private TaskEventProperties properties;
+
+	public EventEmittingJobExecutionListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingJobExecutionListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingJobExecutionListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
 	@Override
 	public void beforeJob(JobExecution jobExecution) {
-		this.messagePublisher.publish(new JobExecutionEvent(jobExecution));
+		this.messagePublisher.publish(properties.getJobExecutionEventBindingName(), new JobExecutionEvent(jobExecution));
 	}
 
 	@Override
 	public void afterJob(JobExecution jobExecution) {
-		this.messagePublisher.publish(new JobExecutionEvent(jobExecution));
+		this.messagePublisher.publish(properties.getJobExecutionEventBindingName(), new JobExecutionEvent(jobExecution));
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -22,8 +22,8 @@ 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.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -43,17 +43,22 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
 
 	private static final Log logger = LogFactory.getLog(EventEmittingSkipListener.class);
 
-	private MessagePublisher<Object> messagePublisher;
+	private final MessagePublisher<Object> messagePublisher;
 
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingSkipListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher<>(output);
+	private TaskEventProperties properties;
+
+	public EventEmittingSkipListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingSkipListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingSkipListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
@@ -62,8 +67,7 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Executing onSkipInRead: " + t.getMessage(), t);
 		}
-		this.messagePublisher.publishWithThrowableHeader("Skipped when reading.",
-				t.getMessage());
+		this.messagePublisher.publishWithThrowableHeader(this.properties.getSkipEventBindingName(), "Skipped when reading.", t.getMessage());
 	}
 
 	@Override
@@ -71,7 +75,7 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Executing onSkipInWrite: " + t.getMessage(), t);
 		}
-		this.messagePublisher.publishWithThrowableHeader(item, t.getMessage());
+		this.messagePublisher.publishWithThrowableHeader(this.properties.getSkipEventBindingName(), item, t.getMessage());
 	}
 
 	@Override
@@ -79,7 +83,7 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Executing onSkipInProcess: " + t.getMessage(), t);
 		}
-		this.messagePublisher.publishWithThrowableHeader(item, t.getMessage());
+		this.messagePublisher.publishWithThrowableHeader(this.properties.getSkipEventBindingName(), item, t.getMessage());
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -21,8 +21,8 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
 import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent;
+import org.springframework.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.core.Ordered;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
@@ -37,28 +37,32 @@ import org.springframework.util.Assert;
 public class EventEmittingStepExecutionListener
 		implements StepExecutionListener, Ordered {
 
-	private MessagePublisher<StepExecutionEvent> messagePublisher;
-
+	private final MessagePublisher<StepExecutionEvent> messagePublisher;
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
-	public EventEmittingStepExecutionListener(MessageChannel output) {
-		Assert.notNull(output, "An output channel is required");
-		this.messagePublisher = new MessagePublisher<>(output);
+	private TaskEventProperties properties;
+
+	public EventEmittingStepExecutionListener(MessagePublisher messagePublisher, TaskEventProperties properties) {
+		Assert.notNull(messagePublisher, "messagePublisher is required");
+		Assert.notNull(properties, "properties is required");
+
+		this.messagePublisher = messagePublisher;
+		this.properties = properties;
 	}
 
-	public EventEmittingStepExecutionListener(MessageChannel output, int order) {
-		this(output);
+	public EventEmittingStepExecutionListener(MessagePublisher messagePublisher, int order, TaskEventProperties properties) {
+		this(messagePublisher, properties);
 		this.order = order;
 	}
 
 	@Override
 	public void beforeStep(StepExecution stepExecution) {
-		this.messagePublisher.publish(new StepExecutionEvent(stepExecution));
+		this.messagePublisher.publish(this.properties.getStepExecutionEventBindingName(), new StepExecutionEvent(stepExecution));
 	}
 
 	@Override
 	public ExitStatus afterStep(StepExecution stepExecution) {
-		this.messagePublisher.publish(new StepExecutionEvent(stepExecution));
+		this.messagePublisher.publish(this.properties.getStepExecutionEventBindingName(), new StepExecutionEvent(stepExecution));
 
 		return stepExecution.getExitStatus();
 	}

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/MessagePublisher.java

@@ -16,8 +16,8 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
+import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -29,31 +29,32 @@ import org.springframework.util.Assert;
  */
 public class MessagePublisher<P> {
 
-	private final MessageChannel listenerEventsChannel;
 
-	public MessagePublisher(MessageChannel listenerEventsChannel) {
-		Assert.notNull(listenerEventsChannel, "listenerEventsChannel must not be null");
-		this.listenerEventsChannel = listenerEventsChannel;
+	private final StreamBridge streamBridge;
+
+	public MessagePublisher(StreamBridge streamBridge) {
+		Assert.notNull(streamBridge, "streamBridge must not be null");
+		this.streamBridge = streamBridge;
 	}
 
-	public final void publish(P payload) {
+	public final void publish(String bindingName, P payload) {
 		if (payload instanceof Message) {
-			this.publishMessage((Message<?>) payload);
+			this.publishMessage(bindingName, (Message<?>) payload);
 		}
 		else {
 			Message<P> message = MessageBuilder.withPayload(payload).build();
-			this.listenerEventsChannel.send(message);
+			this.streamBridge.send(bindingName, message);
 		}
 	}
 
-	private void publishMessage(Message<?> message) {
-		this.listenerEventsChannel.send(message);
+	private void publishMessage(String bindingName, Message<?> message) {
+		this.streamBridge.send(bindingName, message);
 	}
 
-	public void publishWithThrowableHeader(P payload, String header) {
+	public void publishWithThrowableHeader(String bindingName, P payload, String header) {
 		Message<P> message = MessageBuilder.withPayload(payload)
 				.setHeader(BatchJobHeaders.BATCH_EXCEPTION, header).build();
-		publishMessage(message);
+		publishMessage(bindingName, message);
 	}
 
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -67,6 +67,15 @@ public class TaskEventProperties {
 	 */
 	private int skipOrder = Ordered.LOWEST_PRECEDENCE;
 
+	private String jobExecutionEventBindingName = "job-execution-events";
+	private String skipEventBindingName = "skip-events";
+	private String chunkEventBindingName = "chunk-events";
+	private String itemProcessEventBindingName = "item-process-events";
+	private String itemReadEventBindingName = "item-read-events";
+	private String itemWriteEventBindingName = "item-write-events";
+	private String stepExecutionEventBindingName = "step-execution-events";
+	private String taskEventBindingName = "task-events";
+
 	public int getJobExecutionOrder() {
 		return this.jobExecutionOrder;
 	}
@@ -123,4 +132,68 @@ public class TaskEventProperties {
 		this.skipOrder = skipOrder;
 	}
 
+
+	public String getJobExecutionEventBindingName() {
+		return jobExecutionEventBindingName;
+	}
+
+	public void setJobExecutionEventBindingName(String jobExecutionEventBindingName) {
+		this.jobExecutionEventBindingName = jobExecutionEventBindingName;
+	}
+
+	public String getSkipEventBindingName() {
+		return skipEventBindingName;
+	}
+
+	public void setSkipEventBindingName(String skipEventBindingName) {
+		this.skipEventBindingName = skipEventBindingName;
+	}
+
+	public String getChunkEventBindingName() {
+		return chunkEventBindingName;
+	}
+
+	public void setChunkEventBindingName(String chunkEventBindingName) {
+		this.chunkEventBindingName = chunkEventBindingName;
+	}
+
+	public String getItemProcessEventBindingName() {
+		return itemProcessEventBindingName;
+	}
+
+	public void setItemProcessEventBindingName(String itemProcessEventBindingName) {
+		this.itemProcessEventBindingName = itemProcessEventBindingName;
+	}
+
+	public String getItemReadEventBindingName() {
+		return itemReadEventBindingName;
+	}
+
+	public void setItemReadEventBindingName(String itemReadEventBindingName) {
+		this.itemReadEventBindingName = itemReadEventBindingName;
+	}
+
+	public String getItemWriteEventBindingName() {
+		return itemWriteEventBindingName;
+	}
+
+	public void setItemWriteEventBindingName(String itemWriteEventBindingName) {
+		this.itemWriteEventBindingName = itemWriteEventBindingName;
+	}
+
+	public String getStepExecutionEventBindingName() {
+		return stepExecutionEventBindingName;
+	}
+
+	public void setStepExecutionEventBindingName(String stepExecutionEventBindingName) {
+		this.stepExecutionEventBindingName = stepExecutionEventBindingName;
+	}
+
+	public String getTaskEventBindingName() {
+		return taskEventBindingName;
+	}
+
+	public void setTaskEventBindingName(String taskEventBindingName) {
+		this.taskEventBindingName = taskEventBindingName;
+	}
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/launcher/TaskLauncherSink.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.task.launcher;
 
+import java.util.function.Consumer;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -24,10 +26,9 @@ import org.springframework.cloud.deployer.resource.support.DelegatingResourceLoa
 import org.springframework.cloud.deployer.spi.core.AppDefinition;
 import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
 import org.springframework.cloud.deployer.spi.task.TaskLauncher;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.Bean;
 import org.springframework.core.io.Resource;
-import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 /**
@@ -36,7 +37,6 @@ import org.springframework.util.Assert;
  * @author Glenn Renfro
  */
 
-@EnableBinding(Sink.class)
 public class TaskLauncherSink {
 
 	private final static Logger logger = LoggerFactory.getLogger(TaskLauncherSink.class);
@@ -52,13 +52,13 @@ public class TaskLauncherSink {
 
 	/**
 	 * Launches a task upon the receipt of a valid TaskLaunchRequest.
-	 * @param taskLaunchRequest is a TaskLaunchRequest containing the information required
-	 * to launch a task.
-	 * @throws Exception if error occurs during task launch.
+	 * @return the {@link Consumer} that will retrieve messages from binder.
 	 */
-	@ServiceActivator(inputChannel = Sink.INPUT)
-	public void taskLauncherSink(TaskLaunchRequest taskLaunchRequest) throws Exception {
-		launchTask(taskLaunchRequest);
+	@Bean
+	public Consumer<Message<TaskLaunchRequest>> taskLauncherSink() {
+		return messagePayload -> {
+			launchTask(messagePayload.getPayload());
+		};
 	}
 
 	private void launchTask(TaskLaunchRequest taskLaunchRequest) {

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/listener/TaskEventAutoConfiguration.java

@@ -22,22 +22,22 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.cloud.task.batch.listener.support.TaskEventProperties;
 import org.springframework.cloud.task.configuration.SimpleTaskAutoConfiguration;
+import org.springframework.cloud.task.repository.TaskExecution;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
-import org.springframework.integration.gateway.GatewayProxyFactoryBean;
-import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Michael Minella
  * @author Glenn Renfro
  */
 @Configuration(proxyBeanMethods = false)
-@ConditionalOnClass(EnableBinding.class)
+@ConditionalOnClass(StreamBridge.class)
 @ConditionalOnBean(TaskLifecycleListener.class)
 @ConditionalOnExpression("T(org.springframework.util.StringUtils).isEmpty('${spring.batch.job.jobName:}')")
 // @checkstyle:off
@@ -47,38 +47,32 @@ import org.springframework.messaging.MessageChannel;
 @PropertySource("classpath:/org/springframework/cloud/task/application.properties")
 @AutoConfigureBefore(BindingServiceConfiguration.class)
 @AutoConfigureAfter(SimpleTaskAutoConfiguration.class)
+@EnableConfigurationProperties(TaskEventProperties.class)
 public class TaskEventAutoConfiguration {
 
-	/**
-	 * Task Event channels definition.
-	 */
-	public interface TaskEventChannels {
-
-		/**
-		 * Name of the default task events channel.
-		 */
-		String TASK_EVENTS = "task-events";
-
-		@Output(TASK_EVENTS)
-		MessageChannel taskEvents();
-
-	}
-
 	/**
 	 * Configuration for a {@link TaskExecutionListener}.
 	 */
 	@Configuration(proxyBeanMethods = false)
-	@EnableBinding(TaskEventChannels.class)
 	public static class ListenerConfiguration {
-
 		@Bean
-		public GatewayProxyFactoryBean taskEventListener() {
-			GatewayProxyFactoryBean factoryBean = new GatewayProxyFactoryBean(
-					TaskExecutionListener.class);
+		public TaskExecutionListener taskEventEmitter(StreamBridge streamBridge, TaskEventProperties taskEventProperties) {
+			return new TaskExecutionListener() {
+				@Override
+				public void onTaskStartup(TaskExecution taskExecution) {
+					streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);
+				}
 
-			factoryBean.setDefaultRequestChannelName(TaskEventChannels.TASK_EVENTS);
+				@Override
+				public void onTaskEnd(TaskExecution taskExecution) {
+					streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);
+				}
 
-			return factoryBean;
+				@Override
+				public void onTaskFailed(TaskExecution taskExecution, Throwable throwable) {
+					streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);
+				}
+			};
 		}
 
 	}