Add Job Event Messaging to tasks.

* Added test for sample and cleanup

* Added skipEventsListener

* Added new message format.

* Added integration tests.

* Updated application.properties to add json content type

resolves spring-cloud/spring-cloud-task#119

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

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2016 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.springframework.batch.core.ChunkListener;
+import org.springframework.batch.core.ItemProcessListener;
+import org.springframework.batch.core.ItemReadListener;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecutionListener;
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+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.listener.TaskLifecycleListener;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.integration.gateway.GatewayProxyFactoryBean;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * Configures the listeners and channels that are required to emit job messages.
+ * @author Michael Minella
+ * @author Glenn Renfro
+ */
+@Configuration
+@ConditionalOnBean(value = { Job.class, TaskLifecycleListener.class })
+@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events", name = "enabled", havingValue = "true", matchIfMissing = true)
+public class BatchEventAutoConfiguration {
+
+	public final static String JOB_EXECUTION_EVENTS_LISTENER = "jobExecutionEventsListener";
+	public final static String CHUNK_EVENTS_LISTENER = "chunkEventsListener";
+	public final static String STEP_EXECUTION_EVENTS_LISTENER = "stepExecutionEventsListener";
+	public final static String ITEM_READ_EVENTS_LISTENER = "itemReadEventsListener";
+	public final static String ITEM_WRITE_EVENTS_LISTENER = "itemWriteEventsListener";
+	public final static String ITEM_PROCESS_EVENTS_LISTENER = "itemProcessEventsListener";
+	public final static String SKIP_EVENTS_LISTENER = "skipEventsListener";
+
+	@Bean
+	@ConditionalOnMissingBean
+	public TaskBatchEventListenerBeanPostProcessor batchTaskExecutionListenerBeanPostProcessor() {
+		return new TaskBatchEventListenerBeanPostProcessor();
+	}
+
+	@Configuration
+	@EnableBinding(BatchEventsChannels.class)
+	@ConditionalOnMissingBean(name = JOB_EXECUTION_EVENTS_LISTENER)
+	public static class JobExecutionListenerConfiguration {
+
+		@Autowired
+		private BatchEventsChannels listenerChannels;
+
+		@Bean
+		@Lazy
+		public JobExecutionListener jobExecutionEventsListener() {
+			return new EventEmittingJobExecutionListener(listenerChannels.jobExecutionEvents());
+		}
+
+		@Bean
+		public StepExecutionListener stepExecutionEventsListener() {
+			return new EventEmittingStepExecutionListener(listenerChannels.stepExecutionEvents());
+		}
+
+		@Bean
+		@Lazy
+		public GatewayProxyFactoryBean chunkEventsListener() {
+			GatewayProxyFactoryBean factoryBean =
+					new GatewayProxyFactoryBean(ChunkListener.class);
+
+			factoryBean.setDefaultRequestChannel(listenerChannels.chunkEvents());
+
+			return factoryBean;
+		}
+
+		@Bean
+		public ItemReadListener itemReadEventsListener() {
+			return new EventEmittingItemReadEventsListener(listenerChannels.itemReadEvents());
+		}
+
+		@Bean
+		public ItemWriteListener itemWriteEventsListener() {
+			return new EventEmittingItemWriteEventsListener(listenerChannels.itemWriteEvents());
+		}
+
+		@Bean
+		public ItemProcessListener itemProcessEventsListener() {
+			return new EventEmittingItemProcessListener(listenerChannels.itemProcessEvents());
+		}
+
+		@Bean
+		public SkipListener skipEventsListener() {
+			return new EventEmittingSkipListener(listenerChannels.skipEvents());
+		}
+	}
+
+	public interface BatchEventsChannels {
+
+		String JOB_EXECUTION_EVENTS = "job-execution-events";
+		String STEP_EXECUTION_EVENTS = "step-execution-events";
+		String CHUNK_EXECUTION_EVENTS = "chunk-events";
+		String ITEM_READ_EVENTS = "item-read-events";
+		String ITEM_PROCESS_EVENTS = "item-process-events";
+		String ITEM_WRITE_EVENTS = "item-write-events";
+		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();
+
+	}
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2016 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.springframework.batch.core.ItemProcessListener;
+import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Setups up the ItemProcessListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Michael Minella
+ * @author Glenn Renfro
+ */
+public class EventEmittingItemProcessListener implements ItemProcessListener {
+
+	private MessageChannel output;
+	private MessagePublisher<Object> messagePublisher;
+
+	public EventEmittingItemProcessListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher(output);
+	}
+
+	@Override
+	public void beforeProcess(Object item) {
+	}
+
+	@Override
+	public void afterProcess(Object item, Object result) {
+		if (result == null) {
+			messagePublisher.publish("1 item was filtered");
+		}
+		else if (item.equals(result)) {
+			messagePublisher.publish("item equaled result after processing");
+		}
+		else {
+			messagePublisher.publish("item did not equal result after processing");
+		}
+	}
+
+	@Override
+	public void onProcessError(Object item, Exception e) {
+		messagePublisher.publishWithThrowableHeader("Exception while item was being processed", e.getMessage());
+	}
+}

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

@@ -0,0 +1,61 @@
+/*
+ *  Copyright 2016 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemReadListener;
+import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ *  Setups up the ItemReadEventsListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Glenn Renfro
+ */
+public class EventEmittingItemReadEventsListener implements ItemReadListener {
+
+	private static final Logger logger = LoggerFactory.getLogger(EventEmittingItemReadEventsListener.class);
+
+	private MessageChannel output;
+	private MessagePublisher<Object> messagePublisher;
+
+	public EventEmittingItemReadEventsListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher(output);
+	}
+
+	@Override
+	public void beforeRead() {
+
+	}
+
+	@Override
+	public void afterRead(Object item) {
+
+	}
+
+	@Override
+	public void onReadError(Exception ex) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onReadError: " + ex.getMessage(), ex);
+		}
+		this.messagePublisher.publish(ex.getMessage());
+	}
+}

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

@@ -0,0 +1,67 @@
+/*
+ *  Copyright 2016 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 java.util.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ *  Setups up the ItemWriteEventsListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Glenn Renfro
+ */
+public class EventEmittingItemWriteEventsListener implements ItemWriteListener{
+
+	private static final Logger logger = LoggerFactory.getLogger(EventEmittingItemWriteEventsListener.class);
+
+	private MessageChannel output;
+	private MessagePublisher<Object> messagePublisher;
+
+	public EventEmittingItemWriteEventsListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher(output);
+	}
+
+	@Override
+	public void beforeWrite(List items) {
+		messagePublisher.publish(items.size() + " items to be written.");
+	}
+
+	@Override
+	public void afterWrite(List items) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing afterWrite: " + items);
+		}
+		this.messagePublisher.publish(items.size() + " items have been written.");
+	}
+
+	@Override
+	public void onWriteError(Exception exception, List items) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onWriteError: " + exception.getMessage(), exception);
+		}
+		String payload = "Exception while " + items.size() + " items are attempted to be written.";
+		this.messagePublisher.publishWithThrowableHeader(payload, exception.getMessage());
+	}
+}

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

@@ -0,0 +1,52 @@
+/*
+ *  Copyright 2016 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.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.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Setups up the StepExecutionListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Michael Minella
+ * @author Glenn Renfro
+ */
+public class EventEmittingJobExecutionListener implements JobExecutionListener {
+
+	private MessageChannel output;
+
+	private MessagePublisher<JobExecutionEvent> messagePublisher;
+
+	public EventEmittingJobExecutionListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher<>(output);
+	}
+
+	@Override
+	public void beforeJob(JobExecution jobExecution) {
+		this.messagePublisher.publish(new JobExecutionEvent(jobExecution));
+	}
+
+	@Override
+	public void afterJob(JobExecution jobExecution) {
+		this.messagePublisher.publish(new JobExecutionEvent(jobExecution));
+	}
+}

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

@@ -0,0 +1,72 @@
+/*
+ *  Copyright 2016 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+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.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Setups up the SkipProcessListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Glenn Renfro
+ */
+public class EventEmittingSkipListener implements SkipListener {
+
+	private static final Logger logger = LoggerFactory.getLogger(EventEmittingSkipListener.class);
+
+	private MessageChannel output;
+
+	private MessagePublisher<Object> messagePublisher;
+
+	public EventEmittingSkipListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher(output);
+	}
+
+	@Override
+	public void onSkipInRead(Throwable t) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onSkipInRead: " + t.getMessage(), t);
+		}
+		messagePublisher.publishWithThrowableHeader("Skipped when reading.", t.getMessage());
+
+	}
+
+	@Override
+	public void onSkipInWrite(Object item, Throwable t) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onSkipInWrite: " + t.getMessage(), t);
+		}
+		messagePublisher.publishWithThrowableHeader(item, t.getMessage());
+	}
+
+	@Override
+	public void onSkipInProcess(Object item, Throwable t) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Executing onSkipInProcess: " + t.getMessage(), t);
+		}
+		messagePublisher.publishWithThrowableHeader(item, t.getMessage());
+	}
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016 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.springframework.batch.core.ExitStatus;
+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.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Setups up the StepExecutionListener to emit events to the spring cloud stream output channel.
+ *
+ * @author Michael Minella
+ * @author Glenn Renfro
+ */
+public class EventEmittingStepExecutionListener implements StepExecutionListener {
+
+	private MessageChannel output;
+
+	private MessagePublisher<StepExecutionEvent> messagePublisher;
+
+	public EventEmittingStepExecutionListener(MessageChannel output) {
+		Assert.notNull(output, "An output channel is required");
+		this.output = output;
+		this.messagePublisher = new MessagePublisher<>(output);
+	}
+
+	@Override
+	public void beforeStep(StepExecution stepExecution) {
+		this.messagePublisher.publish(new StepExecutionEvent(stepExecution));
+	}
+
+	@Override
+	public ExitStatus afterStep(StepExecution stepExecution) {
+		this.messagePublisher.publish(new StepExecutionEvent(stepExecution));
+
+		return stepExecution.getExitStatus();
+	}
+}

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

@@ -0,0 +1,34 @@
+/*
+ *  Copyright 2016 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;
+
+/**
+ * Headers definitions used by the batch job plugin.
+ *
+ * @author Gunnar Hillert
+ * @since 1.0
+ */
+public final class BatchJobHeaders {
+
+	public static final String BATCH_LISTENER_EVENT_TYPE = "batch_listener_event_type";
+
+	public static final String BATCH_EXCEPTION = "batch_exception";
+
+	private BatchJobHeaders() {
+	}
+
+}

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

@@ -0,0 +1,54 @@
+/*
+ *  Copyright 2016 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;
+
+/**
+ * ExitStatus DTO created so that {@link org.springframework.batch.core.ExitStatus} can be serialized into Json without
+ * having to add mixins to an ObjectMapper
+ * @author Glenn Renfro
+ */
+public class ExitStatus {
+
+	private String exitCode;
+
+	private String exitDescription;
+
+	public ExitStatus(){
+
+	}
+
+	public ExitStatus(String exitCode, String exitDescription) {
+		this.exitCode = exitCode;
+		this.exitDescription = exitDescription;
+	}
+
+	public String getExitCode() {
+		return exitCode;
+	}
+
+	public void setExitCode(String exitCode) {
+		this.exitCode = exitCode;
+	}
+
+	public String getExitDescription() {
+		return exitDescription;
+	}
+
+	public void setExitDescription(String exitDescription) {
+		this.exitDescription = exitDescription;
+	}
+}

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

@@ -0,0 +1,336 @@
+/*
+ *  Copyright 2016 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 java.io.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import org.springframework.batch.core.*;
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * This is a JobEvent DTO created so that a {@link org.springframework.batch.core.JobExecution} can be serialized into
+ * Json without having to add mixins to an ObjectMapper.
+ * @author Glenn Renfro
+ */
+public class JobExecutionEvent extends Entity {
+
+	private JobParametersEvent jobParameters;
+
+	private JobInstanceEvent jobInstance;
+
+	private volatile Collection<StepExecutionEvent> stepExecutions = new CopyOnWriteArraySet<StepExecutionEvent>();
+
+	private volatile BatchStatus status = BatchStatus.STARTING;
+
+	private volatile Date startTime = null;
+
+	private volatile Date createTime = new Date(System.currentTimeMillis());
+
+	private volatile Date endTime = null;
+
+	private volatile Date lastUpdated = null;
+
+	private volatile ExitStatus exitStatus = new ExitStatus("UNKNOWN", null);
+
+	private volatile ExecutionContext executionContext = new ExecutionContext();
+
+	private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
+
+	private String jobConfigurationName;
+
+	public JobExecutionEvent() {
+		super();
+	}
+
+	/**
+	 * Constructor for the StepExecution to initialize the DTO.
+	 *
+	 * @param original the StepExecution to build this DTO around.
+	 */
+	public JobExecutionEvent(JobExecution original) {
+		this.jobParameters = new JobParametersEvent(original.getJobParameters().getParameters());
+		this.jobInstance = new JobInstanceEvent(original.getJobInstance().getId(), original.getJobInstance().getJobName());
+		for(StepExecution stepExecution : original.getStepExecutions()){
+			stepExecutions.add(new StepExecutionEvent(stepExecution));
+		}
+		this.status = original.getStatus();
+		this.startTime = original.getStartTime();
+		this.createTime = original.getCreateTime();
+		this.endTime = original.getEndTime();
+		this.lastUpdated = original.getLastUpdated();
+		this.exitStatus = new ExitStatus(original.getExitStatus().getExitCode(),
+				original.getExitStatus().getExitDescription());
+		this.executionContext = original.getExecutionContext();
+		this.failureExceptions = original.getFailureExceptions();
+		this.jobConfigurationName = original.getJobConfigurationName();
+		this.setId(original.getId());
+		this.setVersion(original.getVersion());
+	}
+
+	public JobParametersEvent getJobParameters() {
+		return this.jobParameters;
+	}
+
+	public Date getEndTime() {
+		return endTime;
+	}
+
+	public void setJobInstance(JobInstanceEvent jobInstance) {
+		this.jobInstance = jobInstance;
+	}
+
+	public void setEndTime(Date endTime) {
+		this.endTime = endTime;
+	}
+
+	public Date getStartTime() {
+		return startTime;
+	}
+
+	public void setStartTime(Date startTime) {
+		this.startTime = startTime;
+	}
+
+	public BatchStatus getStatus() {
+		return status;
+	}
+
+	/**
+	 * Set the value of the status field.
+	 *
+	 * @param status the status to set
+	 */
+	public void setStatus(BatchStatus status) {
+		this.status = status;
+	}
+
+	/**
+	 * Upgrade the status field if the provided value is greater than the
+	 * existing one. Clients using this method to set the status can be sure
+	 * that they don't overwrite a failed status with an successful one.
+	 *
+	 * @param status the new status value
+	 */
+	public void upgradeStatus(BatchStatus status) {
+		this.status = this.status.upgradeTo(status);
+	}
+
+	/**
+	 * Convenience getter for for the id of the enclosing job. Useful for DAO
+	 * implementations.
+	 *
+	 * @return the id of the enclosing job
+	 */
+	public Long getJobId() {
+		if (jobInstance != null) {
+			return jobInstance.getId();
+		}
+		return null;
+	}
+
+	/**
+	 * @param exitStatus
+	 */
+	public void setExitStatus(ExitStatus exitStatus) {
+		this.exitStatus = exitStatus;
+	}
+
+	/**
+	 * @return the exitCode
+	 */
+	public ExitStatus getExitStatus() {
+		return exitStatus;
+	}
+
+	/**
+	 * @return the Job that is executing.
+	 */
+	public JobInstanceEvent getJobInstance() {
+		return jobInstance;
+	}
+
+	/**
+	 * Accessor for the step executions.
+	 *
+	 * @return the step executions that were registered
+	 */
+	public Collection<StepExecutionEvent> getStepExecutions() {
+		return Collections.unmodifiableList(new ArrayList<StepExecutionEvent>(stepExecutions));
+	}
+
+	/**
+	 * Test if this {@link JobExecution} indicates that it is running. It should
+	 * be noted that this does not necessarily mean that it has been persisted
+	 * as such yet.
+	 * @return true if the end time is null
+	 */
+	public boolean isRunning() {
+		return endTime == null;
+	}
+
+	/**
+	 * Test if this {@link JobExecution} indicates that it has been signalled to
+	 * stop.
+	 * @return true if the status is {@link BatchStatus#STOPPING}
+	 */
+	public boolean isStopping() {
+		return status == BatchStatus.STOPPING;
+	}
+
+	/**
+	 * Signal the {@link JobExecution} to stop. Iterates through the associated
+	 * {@link StepExecution}s, calling {@link StepExecution#setTerminateOnly()}.
+	 *
+	 */
+	public void stop() {
+		for (StepExecutionEvent stepExecution : stepExecutions) {
+			stepExecution.setTerminateOnly();
+		}
+		status = BatchStatus.STOPPING;
+	}
+
+	/**
+	 * Sets the {@link ExecutionContext} for this execution
+	 *
+	 * @param executionContext the context
+	 */
+	public void setExecutionContext(ExecutionContext executionContext) {
+		this.executionContext = executionContext;
+	}
+
+	/**
+	 * Returns the {@link ExecutionContext} for this execution. The content is
+	 * expected to be persisted after each step completion (successful or not).
+	 *
+	 * @return the context
+	 */
+	public ExecutionContext getExecutionContext() {
+		return executionContext;
+	}
+
+	/**
+	 * @return the time when this execution was created.
+	 */
+	public Date getCreateTime() {
+		return createTime;
+	}
+
+	/**
+	 * @param createTime creation time of this execution.
+	 */
+	public void setCreateTime(Date createTime) {
+		this.createTime = createTime;
+	}
+
+	public String getJobConfigurationName() {
+		return this.jobConfigurationName;
+	}
+
+	/**
+	 * Package private method for re-constituting the step executions from
+	 * existing instances.
+	 * @param stepExecution
+	 */
+	void addStepExecution(StepExecutionEvent stepExecution) {
+		stepExecutions.add(stepExecution);
+	}
+
+	/**
+	 * Get the date representing the last time this JobExecution was updated in
+	 * the JobRepository.
+	 *
+	 * @return Date representing the last time this JobExecution was updated.
+	 */
+	public Date getLastUpdated() {
+		return lastUpdated;
+	}
+
+	/**
+	 * Set the last time this JobExecution was updated.
+	 *
+	 * @param lastUpdated
+	 */
+	public void setLastUpdated(Date lastUpdated) {
+		this.lastUpdated = lastUpdated;
+	}
+
+	public List<Throwable> getFailureExceptions() {
+		return failureExceptions;
+	}
+
+	/**
+	 * Add the provided throwable to the failure exception list.
+	 *
+	 * @param t
+	 */
+	public synchronized void addFailureException(Throwable t) {
+		this.failureExceptions.add(t);
+	}
+
+	/**
+	 * Return all failure causing exceptions for this JobExecution, including
+	 * step executions.
+	 *
+	 * @return List&lt;Throwable&gt; containing all exceptions causing failure for
+	 * this JobExecution.
+	 */
+	public synchronized List<Throwable> getAllFailureExceptions() {
+
+		Set<Throwable> allExceptions = new HashSet<Throwable>(failureExceptions);
+		for (StepExecutionEvent stepExecution : stepExecutions) {
+			allExceptions.addAll(stepExecution.getFailureExceptions());
+		}
+
+		return new ArrayList<Throwable>(allExceptions);
+	}
+
+	/**
+	 * Deserialize and ensure transient fields are re-instantiated when read
+	 * back
+	 */
+	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
+		stream.defaultReadObject();
+		failureExceptions = new ArrayList<Throwable>();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see org.springframework.batch.core.domain.Entity#toString()
+	 */
+	@Override
+	public String toString() {
+		return super.toString()
+				+ String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s], jobParameters=[%s]",
+				startTime, endTime, lastUpdated, status, exitStatus, jobInstance, jobParameters);
+	}
+
+	/**
+	 * Add some step executions.  For internal use only.
+	 * @param stepExecutions step executions to add to the current list
+	 */
+	public void addStepExecutions(List<StepExecutionEvent> stepExecutions) {
+		if (stepExecutions!=null) {
+			this.stepExecutions.removeAll(stepExecutions);
+			this.stepExecutions.addAll(stepExecutions);
+		}
+	}
+
+
+}

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

@@ -0,0 +1,61 @@
+/*
+ *  Copyright 2016 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.batch.core.Entity;
+import org.springframework.util.Assert;
+
+/**
+ * This is a JobInstance DTO created so that a {@link org.springframework.batch.core.JobInstance} can be serialized into
+ * Json without having to add mixins to an ObjectMapper.
+ *
+ * @author Glenn Renfro
+ */
+
+public class JobInstanceEvent extends Entity implements javax.batch.runtime.JobInstance {
+
+	private String jobName;
+
+	public JobInstanceEvent() {
+
+	}
+
+	public JobInstanceEvent(Long id, String jobName) {
+		super(id);
+		Assert.hasLength(jobName);
+		this.jobName = jobName;
+	}
+
+	/**
+	 * @return the job name. (Equivalent to getJob().getName())
+	 */
+	@Override
+	public String getJobName() {
+		return jobName;
+	}
+
+	@Override
+	public String toString() {
+		return super.toString() + ", Job=[" + jobName + "]";
+	}
+
+	@Override
+	public long getInstanceId() {
+		return super.getId();
+	}
+
+}

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

@@ -0,0 +1,179 @@
+/*
+ *  Copyright 2016 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 java.util.*;
+
+
+/**
+ * This is a JobParameter DTO created so that a {@link org.springframework.batch.core.JobParameter} can be serialized
+ * into Json without having to add mixins to an ObjectMapper.
+ * @author Glenn Renfro
+ */
+public class JobParameterEvent {
+	private Object parameter;
+
+	private JobParameterEvent.ParameterType parameterType;
+
+	private boolean identifying;
+
+	public JobParameterEvent() {
+
+	}
+
+	/**
+	 * Construct a new JobParameter as a String.
+	 */
+	public JobParameterEvent(String parameter, boolean identifying) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.STRING;
+		this.identifying = identifying;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Long.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Long parameter, boolean identifying) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.LONG;
+		this.identifying = identifying;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Date.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Date parameter, boolean identifying) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.DATE;
+		this.identifying = identifying;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Double.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Double parameter, boolean identifying) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.DOUBLE;
+		this.identifying = identifying;
+	}
+
+
+	/**
+	 * Construct a new JobParameter as a String.
+	 */
+	public JobParameterEvent(String parameter) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.STRING;
+		this.identifying = true;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Long.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Long parameter) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.LONG;
+		this.identifying = true;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Date.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Date parameter) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.DATE;
+		this.identifying = true;
+	}
+
+	/**
+	 * Construct a new JobParameter as a Double.
+	 *
+	 * @param parameter
+	 */
+	public JobParameterEvent(Double parameter) {
+		this.parameter = parameter;
+		parameterType = JobParameterEvent.ParameterType.DOUBLE;
+		this.identifying = true;
+	}
+
+	public boolean isIdentifying() {
+		return identifying;
+	}
+
+	/**
+	 * @return the value contained within this JobParameter.
+	 */
+	public Object getValue() {
+
+		if (parameter != null && parameter.getClass().isInstance(Date.class)) {
+			return new Date(((Date) parameter).getTime());
+		}
+		else {
+			return parameter;
+		}
+	}
+
+	/**
+	 * @return a ParameterType representing the type of this parameter.
+	 */
+	public JobParameterEvent.ParameterType getType() {
+		return parameterType;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj instanceof JobParameterEvent == false) {
+			return false;
+		}
+
+		if (this == obj) {
+			return true;
+		}
+
+		JobParameterEvent rhs = (JobParameterEvent) obj;
+		return parameter==null ? rhs.parameter==null && parameterType==rhs.parameterType: parameter.equals(rhs.parameter);
+	}
+
+	@Override
+	public String toString() {
+		return parameter == null ? null : (parameterType == JobParameterEvent.ParameterType.DATE ? "" + ((Date) parameter).getTime()
+				: parameter.toString());
+	}
+
+	@Override
+	public int hashCode() {
+		return 7 + 21 * (parameter == null ? parameterType.hashCode() : parameter.hashCode());
+	}
+
+	/**
+	 * Enumeration representing the type of a JobParameter.
+	 */
+	public enum ParameterType {
+
+		STRING, DATE, LONG, DOUBLE;
+	}
+}

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

@@ -0,0 +1,227 @@
+/*
+ *  Copyright 2016 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 java.util.*;
+
+import org.springframework.batch.core.JobParameter;
+
+/**
+ * This is a JobParametersEvent DTO created so that a {@link org.springframework.batch.core.JobParameters} can be
+ * serialized into Json without having to add mixins to an ObjectMapper.
+ *
+ * @author Glenn Renfro
+ */
+public class JobParametersEvent {
+	private final Map<String,JobParameterEvent> parameters;
+
+	public JobParametersEvent() {
+		this.parameters = new LinkedHashMap<String, JobParameterEvent>();
+	}
+
+	public JobParametersEvent(Map<String,JobParameter> jobParameters) {
+		this.parameters = new LinkedHashMap<String,JobParameterEvent>();
+		for(Map.Entry<String, JobParameter> entry: jobParameters.entrySet()){
+			if(entry.getValue().getValue() instanceof String){
+				parameters.put(entry.getKey(), new JobParameterEvent(((String) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+			}
+			else if(entry.getValue().getValue() instanceof Long){
+				parameters.put(entry.getKey(), new JobParameterEvent(((Long) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+			}
+			else if(entry.getValue().getValue() instanceof Date){
+				parameters.put(entry.getKey(), new JobParameterEvent(((Date) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+			}
+			else if(entry.getValue().getValue() instanceof Double){
+				parameters.put(entry.getKey(), new JobParameterEvent(((Double) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+			}
+		}
+	}
+
+
+	/**
+	 * Typesafe Getter for the Long represented by the provided key.
+	 *
+	 * @param key The key to get a value for
+	 * @return The <code>Long</code> value
+	 */
+	public Long getLong(String key){
+		if (!parameters.containsKey(key)) {
+			return 0L;
+		}
+		Object value = parameters.get(key).getValue();
+		return value==null ? 0L : ((Long)value).longValue();
+	}
+
+	/**
+	 * Typesafe Getter for the Long represented by the provided key.  If the
+	 * key does not exist, the default value will be returned.
+	 *
+	 * @param key to return the value for
+	 * @param defaultValue to return if the value doesn't exist
+	 * @return the parameter represented by the provided key, defaultValue
+	 * otherwise.
+	 */
+	public Long getLong(String key, long defaultValue){
+		if(parameters.containsKey(key)){
+			return getLong(key);
+		}
+		else{
+			return defaultValue;
+		}
+	}
+
+	/**
+	 * Typesafe Getter for the String represented by the provided key.
+	 *
+	 * @param key The key to get a value for
+	 * @return The <code>String</code> value
+	 */
+	public String getString(String key){
+		JobParameterEvent value = parameters.get(key);
+		return value==null ? null : value.toString();
+	}
+
+	/**
+	 * Typesafe Getter for the String represented by the provided key.  If the
+	 * key does not exist, the default value will be returned.
+	 *
+	 * @param key to return the value for
+	 * @param defaultValue to return if the value doesn't exist
+	 * @return the parameter represented by the provided key, defaultValue
+	 * otherwise.
+	 */
+	public String getString(String key, String defaultValue){
+		if(parameters.containsKey(key)){
+			return getString(key);
+		}
+		else{
+			return defaultValue;
+		}
+	}
+
+	/**
+	 * Typesafe Getter for the Long represented by the provided key.
+	 *
+	 * @param key The key to get a value for
+	 * @return The <code>Double</code> value
+	 */
+	public Double getDouble(String key){
+		if (!parameters.containsKey(key)) {
+			return 0.0;
+		}
+		Double value = (Double)parameters.get(key).getValue();
+		return value==null ? 0.0 : value.doubleValue();
+	}
+
+	/**
+	 * Typesafe Getter for the Double represented by the provided key.  If the
+	 * key does not exist, the default value will be returned.
+	 *
+	 * @param key to return the value for
+	 * @param defaultValue to return if the value doesn't exist
+	 * @return the parameter represented by the provided key, defaultValue
+	 * otherwise.
+	 */
+	public Double getDouble(String key, double defaultValue){
+		if(parameters.containsKey(key)){
+			return getDouble(key);
+		}
+		else{
+			return defaultValue;
+		}
+	}
+
+	/**
+	 * Typesafe Getter for the Date represented by the provided key.
+	 *
+	 * @param key The key to get a value for
+	 * @return The <code>java.util.Date</code> value
+	 */
+	public Date getDate(String key){
+		return this.getDate(key,null);
+	}
+
+	/**
+	 * Typesafe Getter for the Date represented by the provided key.  If the
+	 * key does not exist, the default value will be returned.
+	 *
+	 * @param key to return the value for
+	 * @param defaultValue to return if the value doesn't exist
+	 * @return the parameter represented by the provided key, defaultValue
+	 * otherwise.
+	 */
+	public Date getDate(String key, Date defaultValue){
+		if(parameters.containsKey(key)){
+			return (Date)parameters.get(key).getValue();
+		}
+		else{
+			return defaultValue;
+		}
+	}
+
+	/**
+	 * Get a map of all parameters, including string, long, and date.
+	 *
+	 * @return an unmodifiable map containing all parameters.
+	 */
+	public Map<String, JobParameterEvent> getParameters(){
+		return new LinkedHashMap<String, JobParameterEvent>(parameters);
+	}
+
+	/**
+	 * @return true if the parameters is empty, false otherwise.
+	 */
+	public boolean isEmpty(){
+		return parameters.isEmpty();
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if(obj instanceof JobParametersEvent == false){
+			return false;
+		}
+
+		if(obj == this){
+			return true;
+		}
+
+		JobParametersEvent rhs = (JobParametersEvent)obj;
+		return this.parameters.equals(rhs.parameters);
+	}
+
+	@Override
+	public int hashCode() {
+		return 17 + 23 * parameters.hashCode();
+	}
+
+	@Override
+	public String toString() {
+		return parameters.toString();
+	}
+
+	public Properties toProperties() {
+		Properties props = new Properties();
+
+		for (Map.Entry<String, JobParameterEvent> param : parameters.entrySet()) {
+			if(param.getValue() != null) {
+				props.put(param.getKey(), param.getValue().toString());
+			}
+		}
+
+		return props;
+	}
+}

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

@@ -0,0 +1,55 @@
+/*
+ *  Copyright 2016 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.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Utility class that sends batch job listener payloads to the  notification channel.
+ * @author Glenn Renfro
+ */
+public class MessagePublisher<P>{
+	private final MessageChannel listenerEventsChannel;
+
+	public MessagePublisher(MessageChannel listenerEventsChannel) {
+		Assert.notNull(listenerEventsChannel, "listenerEventsChannel must not be null");
+		this.listenerEventsChannel = listenerEventsChannel;
+	}
+
+	public final void publish(P payload) {
+		if (payload instanceof Message) {
+			this.publishMessage((Message<?>) payload);
+		}
+		else {
+			Message<P> message = MessageBuilder.withPayload(payload).build();
+			this.listenerEventsChannel.send(message);
+		}
+	}
+
+	private final void publishMessage(Message<?> message) {
+		this.listenerEventsChannel.send(message);
+	}
+
+	public void publishWithThrowableHeader(P payload, String header) {
+		Message<P> message = MessageBuilder.withPayload(payload).setHeader(BatchJobHeaders.BATCH_EXCEPTION,
+				header).build();
+		publishMessage(message);
+	}
+}

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

@@ -0,0 +1,476 @@
+/*
+ *  Copyright 2016 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 java.io.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Entity;
+
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.util.Assert;
+
+/**
+ * This is a StepExecution DTO created so that a {@link org.springframework.batch.core.StepExecution} can be serialized
+ * into Json without having to add mixins to an ObjectMapper.
+ * @author Glenn Renfro
+ */
+public class StepExecutionEvent extends Entity {
+
+	private volatile long jobExecutionId;
+
+	private volatile String stepName;
+
+	private volatile BatchStatus status = BatchStatus.STARTING;
+
+	private volatile int readCount = 0;
+
+	private volatile int writeCount = 0;
+
+	private volatile int commitCount = 0;
+
+	private volatile int rollbackCount = 0;
+
+	private volatile int readSkipCount = 0;
+
+	private volatile int processSkipCount = 0;
+
+	private volatile int writeSkipCount = 0;
+
+	private volatile Date startTime = new Date(System.currentTimeMillis());
+
+	private volatile Date endTime = null;
+
+	private volatile Date lastUpdated = null;
+
+	private volatile ExecutionContext executionContext = new ExecutionContext();
+
+	private volatile ExitStatus exitStatus = new ExitStatus("EXECUTING", null);
+
+	private volatile boolean terminateOnly;
+
+	private volatile int filterCount;
+
+	private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
+
+
+	public StepExecutionEvent() {
+		super();
+	}
+
+	/**
+	 * Constructor for the StepExecution to initialize the DTO.
+	 *
+	 * @param stepExecution the StepExecution to build this DTO around.
+	 */
+	public StepExecutionEvent(StepExecution stepExecution) {
+		super();
+		Assert.notNull(stepExecution, "StepExecution must be provided to re-hydrate an existing StepExecutionEvent");
+		Assert.notNull(stepExecution.getJobExecution(), "JobExecution must be provided to re-hydrate an existing StepExecutionEvent");
+		setId(stepExecution.getId());
+		this.jobExecutionId = stepExecution.getJobExecutionId();
+		this.stepName = stepExecution.getStepName();
+
+		this.status = stepExecution.getStatus();
+		this.exitStatus = new ExitStatus(stepExecution.getExitStatus().getExitCode(),
+				stepExecution.getExitStatus().getExitDescription());
+		this.executionContext = stepExecution.getExecutionContext();
+		for (Throwable throwable : stepExecution.getFailureExceptions()){
+			this.failureExceptions.add(throwable);
+		}
+		this.terminateOnly = stepExecution.isTerminateOnly();
+
+		this.endTime = stepExecution.getEndTime();
+		this.lastUpdated = stepExecution.getLastUpdated();
+		this.startTime = stepExecution.getStartTime();
+
+		this.commitCount = stepExecution.getCommitCount();
+		this.filterCount = stepExecution.getFilterCount();
+		this.processSkipCount = stepExecution.getProcessSkipCount();
+		this.readCount = stepExecution.getReadCount();
+		this.readSkipCount = stepExecution.getReadSkipCount();
+		this.rollbackCount = stepExecution.getRollbackCount();
+		this.writeCount = stepExecution.getWriteCount();
+		this.writeSkipCount = stepExecution.getWriteSkipCount();
+
+	}
+
+	/**
+	 * Returns the {@link ExecutionContext} for this execution
+	 *
+	 * @return the attributes
+	 */
+	public ExecutionContext getExecutionContext() {
+		return executionContext;
+	}
+
+	/**
+	 * Sets the {@link ExecutionContext} for this execution
+	 *
+	 * @param executionContext the attributes
+	 */
+	public void setExecutionContext(ExecutionContext executionContext) {
+		this.executionContext = executionContext;
+	}
+
+	/**
+	 * Returns the current number of commits for this execution
+	 *
+	 * @return the current number of commits
+	 */
+	public int getCommitCount() {
+		return commitCount;
+	}
+
+	/**
+	 * Sets the current number of commits for this execution
+	 *
+	 * @param commitCount the current number of commits
+	 */
+	public void setCommitCount(int commitCount) {
+		this.commitCount = commitCount;
+	}
+
+	/**
+	 * Returns the time that this execution ended
+	 *
+	 * @return the time that this execution ended
+	 */
+	public Date getEndTime() {
+		return endTime;
+	}
+
+	/**
+	 * Sets the time that this execution ended
+	 *
+	 * @param endTime the time that this execution ended
+	 */
+	public void setEndTime(Date endTime) {
+		this.endTime = endTime;
+	}
+
+	/**
+	 * Returns the current number of items read for this execution
+	 *
+	 * @return the current number of items read for this execution
+	 */
+	public int getReadCount() {
+		return readCount;
+	}
+
+	/**
+	 * Sets the current number of read items for this execution
+	 *
+	 * @param readCount the current number of read items for this execution
+	 */
+	public void setReadCount(int readCount) {
+		this.readCount = readCount;
+	}
+
+	/**
+	 * Returns the current number of items written for this execution
+	 *
+	 * @return the current number of items written for this execution
+	 */
+	public int getWriteCount() {
+		return writeCount;
+	}
+
+	/**
+	 * Sets the current number of written items for this execution
+	 *
+	 * @param writeCount the current number of written items for this execution
+	 */
+	public void setWriteCount(int writeCount) {
+		this.writeCount = writeCount;
+	}
+
+	/**
+	 * Returns the current number of rollbacks for this execution
+	 *
+	 * @return the current number of rollbacks for this execution
+	 */
+	public int getRollbackCount() {
+		return rollbackCount;
+	}
+
+	/**
+	 * Returns the current number of items filtered out of this execution
+	 *
+	 * @return the current number of items filtered out of this execution
+	 */
+	public int getFilterCount() {
+		return filterCount;
+	}
+
+	/**
+	 * Public setter for the number of items filtered out of this execution.
+	 * @param filterCount the number of items filtered out of this execution to
+	 * set
+	 */
+	public void setFilterCount(int filterCount) {
+		this.filterCount = filterCount;
+	}
+
+	/**
+	 * Setter for number of rollbacks for this execution
+	 */
+	public void setRollbackCount(int rollbackCount) {
+		this.rollbackCount = rollbackCount;
+	}
+
+	/**
+	 * Gets the time this execution started
+	 *
+	 * @return the time this execution started
+	 */
+	public Date getStartTime() {
+		return startTime;
+	}
+
+	/**
+	 * Sets the time this execution started
+	 *
+	 * @param startTime the time this execution started
+	 */
+	public void setStartTime(Date startTime) {
+		this.startTime = startTime;
+	}
+
+	/**
+	 * Returns the current status of this step
+	 *
+	 * @return the current status of this step
+	 */
+	public BatchStatus getStatus() {
+		return status;
+	}
+
+	/**
+	 * Sets the current status of this step
+	 *
+	 * @param status the current status of this step
+	 */
+	public void setStatus(BatchStatus status) {
+		this.status = status;
+	}
+
+	/**
+	 * Upgrade the status field if the provided value is greater than the
+	 * existing one. Clients using this method to set the status can be sure
+	 * that they don't overwrite a failed status with an successful one.
+	 *
+	 * @param status the new status value
+	 */
+	public void upgradeStatus(BatchStatus status) {
+		this.status = this.status.upgradeTo(status);
+	}
+
+	public void setStepName(String stepName) {
+		this.stepName = stepName;
+	}
+	/**
+	 * @return the name of the step
+	 */
+	public String getStepName() {
+		return stepName;
+	}
+
+	/**
+	 * @param exitStatus
+	 */
+	public void setExitStatus(ExitStatus exitStatus) {
+		this.exitStatus = exitStatus;
+	}
+
+	/**
+	 * @return the exitCode
+	 */
+	public ExitStatus getExitStatus() {
+		return exitStatus;
+	}
+
+	/**
+	 * On unsuccessful execution after a chunk has rolled back.
+	 */
+	public synchronized void incrementRollbackCount() {
+		rollbackCount++;
+	}
+
+	/**
+	 * @return flag to indicate that an execution should halt
+	 */
+	public boolean isTerminateOnly() {
+		return this.terminateOnly;
+	}
+
+	/**
+	 * Set a flag that will signal to an execution environment that this
+	 * execution (and its surrounding job) wishes to exit.
+	 */
+	public void setTerminateOnly() {
+		this.terminateOnly = true;
+	}
+
+	/**
+	 * @return the total number of items skipped.
+	 */
+	public int getSkipCount() {
+		return readSkipCount + processSkipCount + writeSkipCount;
+	}
+
+	/**
+	 * Increment the number of commits
+	 */
+	public void incrementCommitCount() {
+		commitCount++;
+	}
+
+	/**
+	 * @return the number of records skipped on read
+	 */
+	public int getReadSkipCount() {
+		return readSkipCount;
+	}
+
+	/**
+	 * @return the number of records skipped on write
+	 */
+	public int getWriteSkipCount() {
+		return writeSkipCount;
+	}
+
+	/**
+	 * Set the number of records skipped on read
+	 *
+	 * @param readSkipCount
+	 */
+	public void setReadSkipCount(int readSkipCount) {
+		this.readSkipCount = readSkipCount;
+	}
+
+	/**
+	 * Set the number of records skipped on write
+	 *
+	 * @param writeSkipCount
+	 */
+	public void setWriteSkipCount(int writeSkipCount) {
+		this.writeSkipCount = writeSkipCount;
+	}
+
+	/**
+	 * @return the number of records skipped during processing
+	 */
+	public int getProcessSkipCount() {
+		return processSkipCount;
+	}
+
+	/**
+	 * Set the number of records skipped during processing.
+	 *
+	 * @param processSkipCount
+	 */
+	public void setProcessSkipCount(int processSkipCount) {
+		this.processSkipCount = processSkipCount;
+	}
+
+	/**
+	 * @return the Date representing the last time this execution was persisted.
+	 */
+	public Date getLastUpdated() {
+		return lastUpdated;
+	}
+
+	/**
+	 * Set the time when the StepExecution was last updated before persisting
+	 *
+	 * @param lastUpdated
+	 */
+	public void setLastUpdated(Date lastUpdated) {
+		this.lastUpdated = lastUpdated;
+	}
+
+	public List<Throwable> getFailureExceptions() {
+		return failureExceptions;
+	}
+
+	public void addFailureException(Throwable throwable) {
+		this.failureExceptions.add(throwable);
+	}
+
+	public long getJobExecutionId() {
+		return jobExecutionId;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see
+	 * org.springframework.batch.container.common.domain.Entity#equals(java.
+	 * lang.Object)
+	 */
+	@Override
+	public boolean equals(Object obj) {
+
+		if ( !(obj instanceof StepExecution) || getId() == null) {
+			return super.equals(obj);
+		}
+		StepExecution other = (StepExecution) obj;
+
+		return stepName.equals(other.getStepName()) && (jobExecutionId == other.getJobExecutionId())
+				&& getId().equals(other.getId());
+	}
+
+	/**
+	 * Deserialize and ensure transient fields are re-instantiated when read
+	 * back
+	 */
+	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
+		stream.defaultReadObject();
+		failureExceptions = new ArrayList<Throwable>();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see org.springframework.batch.container.common.domain.Entity#hashCode()
+	 */
+	@Override
+	public int hashCode() {
+		Object jobExecutionId = getJobExecutionId();
+		Long id = getId();
+		return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91
+				* (jobExecutionId != null ? jobExecutionId.hashCode() : 0) + 59 * (id != null ? id.hashCode() : 0);
+	}
+
+	@Override
+	public String toString() {
+		return String.format(getSummary() + ", exitDescription=%s", exitStatus.getExitDescription());
+	}
+
+	public String getSummary() {
+		return super.toString()
+				+ String.format(
+				", name=%s, status=%s, exitStatus=%s, readCount=%d, filterCount=%d, writeCount=%d readSkipCount=%d, writeSkipCount=%d"
+						+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", stepName, status,
+				exitStatus.getExitCode(), readCount, filterCount, writeCount, readSkipCount, writeSkipCount,
+				processSkipCount, commitCount, rollbackCount);
+	}
+}

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

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2016 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 java.lang.reflect.*;
+
+import org.springframework.batch.core.ChunkListener;
+import org.springframework.batch.core.ItemProcessListener;
+import org.springframework.batch.core.ItemReadListener;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.JobExecutionListener;
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.job.AbstractJob;
+import org.springframework.batch.core.step.AbstractStep;
+import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
+import org.springframework.batch.core.step.item.SimpleChunkProcessor;
+import org.springframework.batch.core.step.item.SimpleChunkProvider;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.core.step.tasklet.TaskletStep;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.cloud.task.batch.listener.BatchEventAutoConfiguration;
+import org.springframework.context.ApplicationContext;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * Attaches the listeners to the job and its steps.
+ * Based on the type of bean that is being processed will determine what listener is attached.
+ * <p>
+ *     <ul>
+ *         <li>If the bean is of type AbstactJob then the JobExecutionListener is registered with this bean.</li>
+ *         <li>If the bean is of type AbstactStep then the StepExecutionListener is registered with this bean.</li>
+ *         <li>If the bean is of type TaskletStep then the ChunkEventListener is registered with this bean.</li>
+ *         <li>If the tasklet for the TaskletStep is of type ChunkOrientedTasklet the following listeners will be registered. </li>
+ *         <ul>
+ *             <li>ItemReadListener with the ChunkProvider.</li>
+ *             <li>ItemProcessListener with the ChunkProcessor.</li>
+ *             <li>ItemWriteEventsListener with the ChunkProcessor.</li>
+ *             <li>SkipEventsListener with the ChunkProcessor.</li>
+ *         </ul>
+ *     </ul>
+ * </p>
+ * @author Michael Minella
+ * @author Glenn Renfro
+ */
+public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcessor {
+
+	@Autowired
+	private ApplicationContext applicationContext;
+
+	@Override
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+
+		if (bean instanceof AbstractJob) {
+			JobExecutionListener jobExecutionEventsListener =
+					(JobExecutionListener) this.applicationContext.getBean(
+							BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER);
+
+			AbstractJob job = (AbstractJob) bean;
+			job.registerJobExecutionListener(
+					jobExecutionEventsListener);
+		}
+
+		if (bean instanceof AbstractStep) {
+			StepExecutionListener stepExecutionListener =
+					(StepExecutionListener) this.applicationContext.getBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER);
+
+			AbstractStep step = (AbstractStep) bean;
+			step.registerStepExecutionListener(stepExecutionListener);
+
+			if (bean instanceof TaskletStep) {
+				TaskletStep taskletStep = (TaskletStep) bean;
+				taskletStep.registerChunkListener((ChunkListener) this.applicationContext.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER));
+				Tasklet tasklet = taskletStep.getTasklet();
+
+				if (tasklet instanceof ChunkOrientedTasklet) {
+					Field chunkProviderField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProvider");
+					ReflectionUtils.makeAccessible(chunkProviderField);
+					SimpleChunkProvider chunkProvider = (SimpleChunkProvider) ReflectionUtils.getField(chunkProviderField, tasklet);
+					Field chunkProcessorField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProcessor");
+					ReflectionUtils.makeAccessible(chunkProcessorField);
+					SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils.getField(chunkProcessorField, tasklet);
+					chunkProvider.registerListener((ItemReadListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER));
+					chunkProvider.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
+					chunkProcessor.registerListener((ItemProcessListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER));
+					chunkProcessor.registerListener((ItemWriteListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER));
+					chunkProcessor.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
+				}
+			}
+		}
+
+		return bean;
+	}
+
+	@Override
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		return bean;
+	}
+}

spring-cloud-task-stream/src/main/resources/META-INF/spring.factories

@@ -1 +1,2 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframework.cloud.task.listener.TaskEventAutoConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframework.cloud.task.batch.listener.BatchEventAutoConfiguration

spring-cloud-task-stream/src/main/resources/org/springframework/cloud/task/application.properties

@@ -1 +1,9 @@
 spring.cloud.stream.bindings.task-events.contentType=application/json
+spring.cloud.stream.bindings.item-write-events.contentType=application/json
+spring.cloud.stream.bindings.item-read-events.contentType=application/json
+spring.cloud.stream.bindings.item-process-events.contentType=application/json
+spring.cloud.stream.bindings.skip-events.contentType=application/json
+spring.cloud.stream.bindings.step-execution-events.contentType=application/json
+spring.cloud.stream.bindings.job-execution-events.contentType=application/json
+
+

spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventJobExecutionTests.java

@@ -0,0 +1,111 @@
+/*
+ *  Copyright 2016 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 java.util.*;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInstance;
+import org.springframework.batch.core.JobParameter;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.cloud.task.batch.listener.support.JobExecutionEvent;
+import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * @author Glenn Renfro.
+ */
+public class EventJobExecutionTests {
+
+	private static final String JOB_NAME = "FOODJOB";
+	private static final Long JOB_INSTANCE_ID = 1l;
+	private static final Long JOB_EXECUTION_ID = 2l;
+	private static final String JOB_CONFIGURATION_NAME = "FOO_JOB_CONFIG";
+
+	private JobParameters jobParameters;
+	private JobInstance jobInstance;
+
+	@Before
+	public void setup() {
+		jobInstance = new JobInstance(JOB_INSTANCE_ID, JOB_NAME);
+		jobParameters = new JobParameters();
+	}
+
+	@Test
+	public void testBasic() {
+		JobExecution jobExecution = new JobExecution(jobInstance, JOB_EXECUTION_ID,  jobParameters, JOB_CONFIGURATION_NAME);
+		JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(jobExecution);
+		assertNotNull("jobInstance should not be null", jobExecutionEvent.getJobInstance());
+		assertNotNull("jobParameters should not be null", jobExecutionEvent.getJobParameters());
+		assertEquals("jobConfigurationName did not match expected", JOB_CONFIGURATION_NAME,
+				jobExecutionEvent.getJobConfigurationName());
+
+		assertEquals("jobParameters size did not match", 0, jobExecutionEvent.getJobParameters().getParameters().size());
+		assertEquals("jobInstance name did not match", JOB_NAME, jobExecutionEvent.getJobInstance().getJobName());
+		assertEquals("no step executions were expected", 0, jobExecutionEvent.getStepExecutions().size());
+		assertEquals("exitStatus did not match expected", "UNKNOWN", jobExecutionEvent.getExitStatus().getExitCode());
+	}
+
+	@Test
+	public void testJobParameters() {
+		String[] JOB_PARAM_KEYS = { "A", "B", "C", "D" };
+		Date testDate = new Date();
+		JobParameter[] PARAMETERS = { new JobParameter("FOO", true), new JobParameter(1L, true),
+				new JobParameter(1D, true), new JobParameter(testDate, false) };
+
+		Map jobParamMap = new LinkedHashMap<String, JobParameter>();
+		for (int paramCount = 0; paramCount < JOB_PARAM_KEYS.length; paramCount++) {
+			jobParamMap.put(JOB_PARAM_KEYS[paramCount], PARAMETERS[paramCount]);
+		}
+		jobParameters = new JobParameters(jobParamMap);
+		JobExecution jobExecution = new JobExecution(jobInstance, JOB_EXECUTION_ID,  jobParameters, JOB_CONFIGURATION_NAME);
+		JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(jobExecution);
+
+		assertNotNull("Job Parameter A was expected", jobExecutionEvent.getJobParameters().getString("A"));
+		assertNotNull("Job Parameter B was expected", jobExecutionEvent.getJobParameters().getLong("B"));
+		assertNotNull("Job Parameter C was expected", jobExecutionEvent.getJobParameters().getDouble("C"));
+		assertNotNull("Job Parameter D was expected", jobExecutionEvent.getJobParameters().getDate("D"));
+
+		assertEquals("Job Parameter A value was not correct", "FOO", jobExecutionEvent.getJobParameters().getString("A"));
+		assertEquals("Job Parameter B value was not correct", new Long(1), jobExecutionEvent.getJobParameters().getLong("B"));
+		assertEquals("Job Parameter C value was not correct", new Double(1), jobExecutionEvent.getJobParameters().getDouble("C"));
+		assertEquals("Job Parameter D value was not correct", testDate, jobExecutionEvent.getJobParameters().getDate("D"));
+	}
+
+	@Test
+	public void testStepExecutions() {
+		JobExecution jobExecution = new JobExecution(jobInstance, JOB_EXECUTION_ID,  jobParameters, JOB_CONFIGURATION_NAME);
+		List<StepExecution> stepsExecutions = new ArrayList<>();
+		stepsExecutions.add( new StepExecution("foo", jobExecution));
+		stepsExecutions.add( new StepExecution("bar", jobExecution));
+		stepsExecutions.add( new StepExecution("baz", jobExecution));
+		jobExecution.addStepExecutions(stepsExecutions);
+
+		JobExecutionEvent jobExecutionsEvent = new JobExecutionEvent(jobExecution);
+		assertEquals("stepExecutions count is incorrect", 3, jobExecutionsEvent.getStepExecutions().size());
+		Iterator<StepExecutionEvent> iter = jobExecutionsEvent.getStepExecutions().iterator();
+		assertEquals("foo stepExecution is not present", "foo", iter.next().getStepName());
+		assertEquals("bar stepExecution is not present", "bar", iter.next().getStepName());
+		assertEquals("baz stepExecution is not present", "baz", iter.next().getStepName());
+
+	}
+}

spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventStepExecutionTests.java

@@ -0,0 +1,63 @@
+/*
+ *  Copyright 2016 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.junit.Test;
+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.cloud.task.batch.listener.support.StepExecutionEvent;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Glenn Renfro
+ */
+public class EventStepExecutionTests {
+	private static final String JOB_NAME = "FOO_JOB";
+	private static final String STEP_NAME = "STEP_NAME";
+	private static final Long JOB_INSTANCE_ID = 1l;
+	private static final Long JOB_EXECUTION_ID = 2l;
+	private static final String JOB_CONFIGURATION_NAME = "FOO_JOB_CONFIG";
+
+	@Test
+	public void testBasic(){
+		JobInstance jobInstance = new JobInstance(JOB_INSTANCE_ID, JOB_NAME);
+		JobParameters jobParameters = new JobParameters();
+		JobExecution jobExecution = new JobExecution(jobInstance, JOB_EXECUTION_ID,  jobParameters, JOB_CONFIGURATION_NAME);
+
+		StepExecution stepExecution = new StepExecution(STEP_NAME, jobExecution);
+		stepExecution.setCommitCount(1);
+		stepExecution.setReadCount(2);
+		stepExecution.setWriteCount(3);
+		stepExecution.setReadSkipCount(4);
+		stepExecution.setWriteSkipCount(5);
+
+		StepExecutionEvent stepExecutionEvent = new StepExecutionEvent(stepExecution);
+		assertEquals("stepName result was not as expected", STEP_NAME, stepExecutionEvent.getStepName());
+		assertEquals("startTime result was not as expected", stepExecution.getStartTime(), stepExecutionEvent.getStartTime());
+		assertEquals("endTime result was not as expected", stepExecution.getEndTime(), stepExecutionEvent.getEndTime());
+		assertEquals("lastUpdated result was not as expected", stepExecution.getLastUpdated(), stepExecutionEvent.getLastUpdated());
+		assertEquals("commitCount result was not as expected", stepExecution.getCommitCount(), stepExecutionEvent.getCommitCount());
+		assertEquals("readCount result was not as expected", stepExecution.getReadCount(), stepExecutionEvent.getReadCount());
+		assertEquals("readSkipCount result was not as expected", stepExecution.getReadSkipCount(), stepExecutionEvent.getReadSkipCount());
+		assertEquals("writeCount result was not as expected", stepExecution.getWriteCount(), stepExecutionEvent.getWriteCount());
+		assertEquals("writeSkipCount result was not as expected", stepExecution.getWriteSkipCount(), stepExecutionEvent.getWriteSkipCount());
+		assertEquals("skipCount result was not as expected", stepExecution.getSkipCount(), stepExecutionEvent.getSkipCount());
+	}
+}

spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/launcher/TaskLauncherSinkTests.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.task.launcher;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -27,6 +28,7 @@ import org.springframework.boot.test.SpringApplicationConfiguration;
 import org.springframework.cloud.deployer.spi.task.LaunchState;
 import org.springframework.cloud.stream.annotation.Bindings;
 import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
 import org.springframework.cloud.task.launcher.configuration.TaskConfiguration;
 import org.springframework.cloud.task.launcher.util.TaskLauncherSinkApplication;
 import org.springframework.context.ApplicationContext;
@@ -39,6 +41,9 @@ import static org.junit.Assert.assertEquals;
 @SpringApplicationConfiguration(classes = {TaskLauncherSinkApplication.class, TaskConfiguration.class} )
 public class TaskLauncherSinkTests {
 
+	@ClassRule
+	public static RedisTestSupport redisTestSupport = new RedisTestSupport();
+
 	private final static String DEFAULT_STATUS = "test_status";
 
 	@Autowired

spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/listener/TaskEventTests.java

@@ -15,11 +15,12 @@
  */
 package org.springframework.cloud.task.listener;
 
+import org.junit.Rule;
 import org.junit.Test;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
-import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
+import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
 import org.springframework.cloud.task.configuration.EnableTask;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Configuration;
@@ -28,19 +29,24 @@ import static org.junit.Assert.assertNotNull;
 
 /**
  * @author Michael Minella
+ * @author Ilayaperumal Gopinathan
  */
 public class TaskEventTests {
 
+	@Rule
+	public RedisTestSupport redisTestSupport = new RedisTestSupport();
+
+	private static final String TASK_NAME = "taskEventTest";
+
 	@Test
 	public void testDefaultConfiguration() {
 		ConfigurableApplicationContext applicationContext =
-				SpringApplication.run(new Object[] {TaskEventsConfiguration.class,
+				SpringApplication.run(new Object[]{ TaskEventsConfiguration.class,
 								TaskEventAutoConfiguration.class,
-								PropertyPlaceholderAutoConfiguration.class,
-								TestSupportBinderAutoConfiguration.class},
-						new String[] {"--spring.cloud.task.closecontext.enable=false",
+								PropertyPlaceholderAutoConfiguration.class },
+						new String[]{ "--spring.cloud.task.closecontext.enable=false",
 								"--spring.main.web-environment=false",
-								"--spring.cloud.stream.defaultBinder=test"});
+								"--spring.cloud.stream.defaultBinder=test" });
 
 		assertNotNull(applicationContext.getBean("taskEventListener"));
 		assertNotNull(applicationContext.getBean(TaskEventAutoConfiguration.TaskEventChannels.class));
@@ -50,4 +56,5 @@ public class TaskEventTests {
 	@EnableTask
 	public static class TaskEventsConfiguration {
 	}
+
 }