Streams

Introduction

In Spring XD, a basic stream defines the ingestion of data from a source to a sink. Streams are managed using the XD Admin Server and defined in terms of source and sink modules. Modules are predefined configurations for sources and sinks and examples can be found in the modules directory within the spring-xd project ^[1]. When you ask the server to set up a new stream, it looks in the modules directory for the source and sink definitions.

The XD Admin Server

The XD Admin server exposes a REST API (at the /streams) endpoint which allows you to create new streams ^[2]. It uses Redis for TODO brief list???, so make sure Redis is running. You can then run ./gradlew spring-xd-dirt:run from the root directory to start the server. New streams are created by posting stream definitions to the endpoint. The defintions are built from a simple DSL. For example, let’s walk through what happens if we execute the following request

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

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 ^[3]. 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

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 REST API which the server uses to create streams. The http source accepts data on a different port (default 9000), from the API one.

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

$ curl -X POST -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 -X POST -d "hello" http://localhost:9000
$ curl -X POST -d "goodbye" http://localhost:9000

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

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.

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 -X POST -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.

---

1. Using the filesystem is just one possible way of configuring a module registry. A Redis implementation would be another option.

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

3. The definitions are just standard Spring configuration files