spring-cloud-task/spring-cloud-task-docs/src/main/asciidoc/stream.adoc

[[stream-integration]]
= Spring Cloud Stream Integration

[[partintro]]
--
A task by itself can be useful, but it's the integration of a task into a larger ecosystem
that allows it to be useful for more complex processing and orchestration.  This section
covers the integration options for Spring Cloud Task and Spring Cloud Stream.
--

[[stream-integration-launching-sink]]
== Launching a task from a Spring Cloud Stream

Allows a user to launch tasks from a stream.  This is done by creating a sink that
listens for a message that contains a `TaskLaunchRequest` as its payload.  The
TaskLaunchRequest contains the maven coordinates to the Task that is to be executed.  It
also has a Map that contains the environment variables that will be used by the Task.

NOTE: If the payload is of a different type then the sink will throw an exception.

For example  a stream can be created that has a processor that takes in data from a
http source and creates a `GenericMessage` that contains the `TaskLaunchRequest` and sends
the message to its output channel. The task sink would then receive the message from its
input channnel and then launch the task.

To create a taskSink a user needs to only create a spring boot app that includes the
following annotation `EnableTaskLauncher`.  The code would look something like this:

```
@SpringBootApplication
@EnableTaskLauncher
public class TaskSinkApplication {
	public static void main(String[] args) {
		SpringApplication.run(TaskSinkApplication.class, args);
	}
}
```

A sample Sink and Processor have been made available to you in the samples module
of the Spring Cloud Task project.  To install these samples into your local maven
repository execute a maven build from the `spring-cloud-task-samples` directory with the
property `skipInstall` set to false.  For example:
`mvn clean install`.

NOTE:  The maven.remoteRepositories.springRepo.url property will need to be set to
the location of the remote repository from which the über-jar is located.  If not
set, then there will be no remote repository, so it will rely upon the local repository only.

[[stream-integration-launching-sink-dataflow]]
=== Spring Cloud Data Flow

To create a stream in Spring Cloud Data Flow first we would want to register the Task Sink
Application we created.  In the example below we are registering the Processor and Sink
sample applications using the Spring Cloud Data Flow shell:

```
app register --name taskSink --type sink --uri maven://io.spring.cloud:tasksink:<version>
app register --name taskProcessor --type processor --uri maven:io.spring.cloud:taskprocessor:<version>
```

Creating a stream from the Spring Cloud Data Flow shell would look like this:

```
stream create foo --definition "http --server.port=9000|taskProcessor|taskSink" --deploy
```

[[stream-integration-events]]
== Spring Cloud Task Events

Spring Cloud Task provides the ability to emit events via Spring Cloud Stream channel
when the task is executed via a Spring Cloud Stream channel.  A task listener is used to
publish the `TaskExecution` on a message channel named `task-events`.  This feature is
autowired into any task that has `spring-cloud-stream` on its classpath in addition to the
`spring-cloud-stream` and a task defined.

NOTE: To disable the event emitting listener, set the property
`spring.cloud.task.events.enabled` to `false`.

With the appropriate classpath defined, a simple task like this:

```
@SpringBootApplication
@EnableTask
public class TaskEventsApplication {

	public static void main(String[] args) {
		SpringApplication.run(TaskEventsApplication.class, args);
	}

	@Configuration
	public static class TaskConfiguration {

		@Bean
		public CommandLineRunner commandLineRunner() {
			return new CommandLineRunner() {
				@Override
				public void run(String... args) throws Exception {
					System.out.println("The CommandLineRunner was executed");
				}
			};
		}
	}
}
```

will emit the `TaskExecution` as an event on the `task-events` channel (both at the start
and end of the task).

NOTE: Configuration of the content type may be required via
`--spring.cloud.stream.bindings.task-events.contentType=<CONTENT_TYPE>` if the processor
or sink downstream does not have the spring-cloud-task-core jar on its classpath.

NOTE: A binder implementation is also required to be on the classpath.

[[stream-integration-disable-task-events]]
=== Disabling Specific Task Events

To task events, the `spring.cloud.task.events.enabled` property can be set to `false`.


[[stream-integration-batch-events]]
== Spring Batch Events

When executing a Spring Batch job via a task, Spring Cloud Task can be configured to emit
informational messages based on the Spring Batch listeners available in Spring Batch.
Specifically the following Spring Batch listeners are autoconfigured into each batch job and
emit messages on the associated Spring Cloud Stream channels when run via Spring Cloud
Task:

* `JobExecutionListener` - `job-execution-events`
* `StepExecutionListener` - `step-execution-events`
* `ChunkListener` - `chunk-events`
* `ItemReadListener` - `item-read-events`
* `ItemProcessListener` - `item-process-events`
* `ItemWriteListener` - `item-write-events`
* `SkipListener` - `skip-events`

The above listeners are autoconfigured into any `AbstractJob` when the appropriate
beans exist in the context (a `Job` and a `TaskLifecycleListener`). Configuration to
listen to these events is handled the same way binding to any other Spring
Cloud Stream channel is done.  Our task (the one running the batch job) serves as a
`Source`, with the listening applications serving as either a `Processor` or `Sink`.

An example could be to have an application listening to the `job-execution-events` channel
for the start and stop of a job.  To configure the listening application, you'd configure
the input to be `job-execution-events` as follows

```
spring.cloud.stream.bindings.input.destination=job-execution-events
```

NOTE: A binder implementation is also required to be on the classpath.

=== Sending Batch Events to different channels

One of the options that Spring Cloud Task offers for batch events is the ability to alter the channel to which a
specific listener can emit its messages.   To do this use the following configuration:
`spring.cloud.stream.bindings.<the channel>.destination=<new destination>`.
For example: If StepExecutionListener needs to emit its messages to another channel `my-step-execution-events`
instead of the default `step-execution-events` the following configuration can be added:

```
spring.cloud.stream.bindings.step-execution-events.destination=my-step-execution-events`
```

=== Disabling Batch Events
To disable the all batch event listener functionality, use the following configuration:

```
spring.cloud.task.batch.events.enabled=false
```

To disable a specific batch event use the following configuration:
`spring.cloud.task.events.<batch event listener>.enabled=false`:

```
spring.cloud.task.batch.events.job-execution.enabled=false
spring.cloud.task.batch.events.step-execution.enabled=false
spring.cloud.task.batch.events.chunk.enabled=false
spring.cloud.task.batch.events.item-read.enabled=false
spring.cloud.task.batch.events.item-process.enabled=false
spring.cloud.task.batch.events.item-write.enabled=false
spring.cloud.task.batch.events.skip.enabled=false
```