Skip to content

Streams

Jump to bottom Edit New page
Luke Taylor edited this page May 23, 2013 · 16 revisions

Introduction

In Spring XD, a basic stream defines the ingestion of data from a source to a sink. Streams are managed using the StreamServer from the spring-xd-dirt subproject 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 StreamServer to set up a new stream, it looks in the modules directory for the source and sink definitions.

The StreamServer

The StreamServer is a prototype implementation for managing streams [2]. You can start it by running ./gradlew launch from the root directory. It exposes a REST API (at the /streams) endpoint to which you can post commands to define new streams. The commands are built from a simple DSL. For example, let’s walk through what happens if we execute the following request

$ curl -X POST -d "testsource | testsink" http://localhost:8080/streams/mystream

The DSL uses the "pipe" symbol |, to connect a source to a stream. The stream server finds the testsource and testsink definitions in the modules directory and uses them to setup the stream [3]. In this simple example, the testsource simply sends a single "hello" message, and the testsink outputs it using the logging framework.

$ processing module 'Module [name=testsink, type=sink]' from group 'example' with index: 1
$ processing module 'Module [name=testsource, type=source]' from group 'example' with index: 0
$ 16:11:50,542  WARN ThreadPoolTaskScheduler-1 handler.LoggingHandler:141 - hello

Other Source and Sink Types

DSL Syntax

In the example above, we connected a source directly to a sink. It is also possible to interperse "processors" between the two. TODO: Example ??? Mark P. says use taps instead.

1. Using the filesystem is just one possible way of configuring a module registry. A Redis implementation would be another option
2. The server name is likely to change and the functionality will be expanded beyond basic stream handling
3. The definitions are just standard Spring configuration files

Clone this wiki locally