Streams

Introduction

In Spring XD, a basic stream defines the ingestion of event driven data from a source to a sink that passes through any number of processors. Stream processing is performed inside the XD Containers and the deployment of stream definitions to containers is done via the XD Admin Server. The Getting Started section shows you how to start these servers.

Sources, sinks and processors are predefined configurations of a module. Module definitions are found in the xd/modules directory. ^[1]. Modules definitions are standard Spring configuration files that use existing Spring classes, such as Input/Output adapters and Transformers from Spring Integration that support general Enterprise Integration Patterns.

A high level DSL is used to create stream definitions. The DSL to define a stream that has an http source and a file sink (with no processors) is shown below

http | file

The DSL mimics a UNIX pipes and filters syntax. Default values for ports and filenames are used in this example but can be overriden using -- options, such as

http --port 8091 | file --dir=/tmp/httpdata/

To create these stream definitions you make an HTTP POST request to the XD Admin Server. More details can be found in the sections below.

Creating a Simple Stream

The XD Admin server in the M1 release exposes a simple POST endpoint (located at http://host:8080/streams) which allows you to create new streams. ^[2]. A full RESTful API for managing the lifecycle of stream definitions will be provided in the M2 release.

New streams are created by posting stream definitions. The definitions are built from a simple DSL. For example, let’s walk through what happens if we execute the following request

$ curl -d "time | log" http://localhost:8080/streams/ticktock

This defines a stream named ticktock based off the DSL expression time | log. The DSL uses the "pipe" symbol |, to connect a source to a sink. The stream server finds the time and log definitions in the modules directory and uses them to setup the stream. In this simple example, the time source simply sends the current time as a message each second, and the log sink outputs it using the logging framework.

processing module 'Module [name=log, type=sink]' from group 'ticktock' with index: 1
processing module 'Module [name=time, type=source]' from group 'ticktock' with index: 0
17:26:18,774  WARN ThreadPoolTaskScheduler-1 logger.ticktock:141 - Thu May 23 17:26:18 EDT 2013

Deleting a Stream

You can delete a stream by sending an HTTP DELETE request to the original URL you used to create the stream:

$ curl -X DELETE http://localhost:8080/streams/ticktock

Other Source and Sink Types

Let’s try something a bit more complicated and swap out the time source for something else. Another supported source type is http, which accepts data for ingestion over HTTP POSTs. Note that this should not be confused with the POST requests to the Admin Server used to create streams. The http source accepts data on a different port (default 9000), from the Admin Server (default 8080).

To create a stream using an http source, but still using the same log sink, we would change the original command above to

$ curl -d "http | log" http://localhost:8080/streams/myhttpteststream

which will produce the following output from the server

processing module 'Module [name=log, type=sink]' from group 'myhttpteststream' with index: 1
processing module 'Module [name=http, type=source]' from group 'myhttpteststream' with index: 0

Note that we don’t see any other output this time until we actually post some data

$ curl -d "hello" http://localhost:9000
$ curl -d "goodbye" http://localhost:9000

and the stream will then funnel the data from the http source to the output log implemented by the log sink

15:08:01,676  WARN ThreadPoolTaskScheduler-1 logger.myhttpteststream:141 - hello
15:08:12,520  WARN ThreadPoolTaskScheduler-1 logger.myhttpteststream:141 - goodbye

Of course, we could also change the sink implementation. You could pipe the output to a file (file), to hadoop (hdfs) or to any of the other sink modules which are provided. You can also define your own modules.

Simple Stream Processing

As an example of a simple processing step, we can transform the payload of the HTTP posted data to upper case using the stream definitions

http | transform --expression=payload.toUpperCase() | log

To create this stream sent a POST request to the Admin Server

$ curl -d "http | transform --expression=payload.toUpperCase() | log" http://localhost:8080/streams/myprocstream

Posting some data

$ curl -d "hello" http://localhost:9000

Will result in an uppercased 'hello' in the log

15:18:21,345  WARN ThreadPoolTaskScheduler-1 logger.myhttpteststream:141 - HELLO

See the Processors section for more information.

DSL Syntax

In the examples above, we connected a source to a sink using the pipe symbol |. You can also pass parameters to the source and sink configurations. The parameter names will depend on the individual module implementations, but as an example, the http source module exposes a port setting which allows you to change the data ingestion port from the default value. To create the stream using port 8000, we would use

$ curl -d "http --port=8000 | log" http://localhost:8080/streams/myhttpteststream

If you know a bit about Spring configuration files, you can inspect the module definition to see which properties it exposes. Alternatively, you can read more in the source and sink documentation.

The Spring XD M2 release will provide a DSL for non-linear flows, e.g. a directed graph.

---

1. Using the filesystem is just one possible way of storing module defintions. Other backends will be supported in the future, e.g. Redis.

2. The server is implemented by the AdminMain class in the spring-xd-dirt subproject