Sources

Introduction

In this section we will show some variations on input sources. As a prerequisite start the XD Container as instructed in the Getting Started page.

The Sources covered are

HTTP
Tail
Twitter Search
Gemfire CQ
Syslog
TCP

Future releases will provide support for RabbitMQ, JMS, and other currently available Spring Integration Adatpers. For information on how to adapt an existing Spring Integration Adapter for use in Spring XD see the section Creating a Source Module.

HTTP

To create a stream definition in the server post using curl

$ curl -X POST -d "http | file" http://localhost:8080/streams/httptest

Make sure the default output directory exists

$ mkdir -p /tmp/xd/output/

Post some data to the http server on the default port of 9000

$ curl -X POST -d "hello world" http://localhost:9000

See if the data ended up in the file

$ cat /tmp/xd/output/httptest

HTTP with options

The http source has one option

port: The http port where data will be posted (default: 9000)

Here is an example

$ curl -X POST -d "http --port=9020 | file" http://localhost:8080/streams/httptest9020

$ curl -X POST -d "hello world" http://localhost:9020

$ cat /tmp/xd/output/httptest9020

Tail

Make sure the default input directory exists

$ mkdir -p /tmp/xd/input

Create an empty file to tail (this is not needed on some platforms such as Linux)

touch /tmp/xd/input/tailtest

To create a stream definition post using curl

$ curl -X POST -d "tail | file" http://localhost:8080/streams/tailtest

Send some text into the file being monitored

$ echo blah >> /tmp/xd/input/tailtest

See if the data ended up in the file

$ cat /tmp/xd/output/tailtest

Tail with options

The tail source has 3 options:

name: the absolute path to the file to tail (default: /tmp/xd/input/<streamName>)
lines: the number of lines from the end of an existing file to tail (default: 0)
delay: on platforms that don’t wait for a missing file to appear, how often (ms) to look for the file (default: 5000)

Here is an example

$ curl -X POST -d "tail --name=/tmp/foo --lines=5 | file --name=bar" http://localhost:8080/streams/tailtest

$ echo blah >> /tmp/foo

$ cat /tmp/xd/output/bar

Tail Status Events

Some platforms, such as linux, send status messages to stderr. The tail module sends these events to a logging adapter, at WARN level; for example…

[message=tail: cannot open `/tmp/xd/input/tailtest' for reading: No such file or directory, file=/tmp/xd/input/tailtest]
[message=tail: `/tmp/xd/input/tailtest' has become accessible, file=/tmp/xd/input/tailtest]

Twitter Search

The twittersearch source has one option

query: The query that will be run against Twitter (required)

To create a stream definition in the server post using curl

$ curl -X POST -d "twittersearch --query='#springone2gx' | file" http://localhost:8080/streams/springone2gx

Make sure the default output directory for the file sink exists

$ mkdir -p /tmp/xd/output/

Let the twittersearch run for a little while and then check to see if some data ended up in the file

$ cat /tmp/xd/output/springone2gx

GemFire Continuous Query (CQ)

Continuous query allows client applications to create a GemFire query using Object Query Language(OQL) and register a CQ listener which subscribes to the query and is notified every time the query 's result set changes. The gemfire_cq source registers a CQ which will post CQEvent messages to the stream.

Launching the XD GemFire Server

This source requires a cache server to be running in a separate process and its host and port must be known (NOTE: GemFire locators are not supported yet). The XD distribution includes a GemFire server executable suitable for development and test purposes. This is a Java main class that runs with a Spring configured cache server. The configuration is passed as a command line argument to the server’s main method. The configuration includes a cache server port and one or more configured region. XD includes a sample cache configuration called cq-demo. This starts a server on port 40404 and creates a region named Stocks. A Logging cache listener is configured for the region to log region events. (TBD: describe launch script)

Options

The qemfire-cq source has the following options

query: The query string in Object Query Language(OQL) (required, String)
gemfireHost: The host on which the GemFire server is running. (default: localhost)
gemfirePort: The port on which the GemFire server is running. (default: 40404)

Here is an example. Create two streams: One to write http messages to a Gemfire region named Stocks, and another to execute the CQ.

$ curl -X POST -d "http --port=9090 | gemfire-json-server --regionName=Stocks" --keyExpression=payload.getField('symbol')" http://localhost:8080/streams/stocks
$ curl -X POST -d "gemfire-cq --query=Select * from /Stocks where symbol='VMW' | file" http://localhost:8080/streams/cqtest

Now send some messages to the stocks stream.

$ curl -X POST -d "{\"symbol\":\"VMW\", \"price\":73}" http:localhost:9090
$ curl -X POST -d "{\"symbol\":\"VMW\", \"price\":78}" http:localhost:9090
$ curl -X POST -d "{\"symbol\":\"VMW\", \"price\":80}" http:localhost:9090

The cqtest stream is now listening for any stock quote updates for VMW. Presumably, another process is updating the cache. You may create a separate stream to test this (see GemfireServer for instructions).

As updates are posted to the cache you should see them captured in the output file:

$cat /tmp/xd/output/cqtest

CqEvent [CqName=GfCq1; base operation=CREATE; cq operation=CREATE; key=VMW; value=PDX[1,__GEMFIRE_JSON]{price=73, symbol=VMW}]
CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=UPDATE; key=VMW; value=PDX[1,__GEMFIRE_JSON]{price=78, symbol=VMW}]
CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=UPDATE; key=VMW; value=PDX[2,__GEMFIRE_JSON]{price=80, symbol=VMW}]

Syslog

Two syslog sources are provided: syslog-udp and syslog-tcp. They both support the following options:

port: the port on which the system will listen for syslog messages (default: 11111)

To create a stream definition post using curl

$ curl -X POST -d "syslog-udp --port=1514 | file" http://localhost:8080/streams/syslogtest

or

$ curl -X POST -d "syslog-tcp --port=1514 | file" http://localhost:8080/streams/syslogtest

Send a test message to the syslog

logger -p local3.info -t TESTING "Test Syslog Message"

See if the data ended up in the file

$ cat /tmp/xd/output/syslogtest

Refer to your syslog documentation to configure the syslog daemon to forward syslog messages to the stream; some examples are:

UDP - Mac OSX (syslog.conf) and Ubuntu (rsyslog.conf)

*.*	@localhost:11111

TCP - Ubuntu (rsyslog.conf)

$ModLoad omfwd
*.*	@@localhost:11111

Restart the syslog daemon after reconfiguring.

TCP

To create a stream definition in the server, post using curl

$ curl -X POST -d "tcp | file" http://localhost:8080/streams/tcptest

This will create the default TCP source and send data read from it to the tcptest file.

TCP is a streaming protocol and some mechanism is needed to frame messages on the wire. A number of decoders are available, the default being 'CRLF' which is compatible with Telnet.

$ telnet localhost 1234
Trying ::1...
Connected to localhost.
Escape character is '^]'.
foo
^]

telnet> quit
Connection closed.

See if the data ended up in the file

$ cat /tmp/xd/output/tcptest

TCP with options

The TCP source has the following options

port: the port on which to listen (default: 1234)
reverse-lookup: perform a reverse DNS lookup on the remote IP Address (default: false)
socket-timeout: the timeout (ms) before closing the socket when no data received (default: 120000)
nio: whether or not to use NIO. NIO is more efficient when there are many connections. (default: false)
decoder: how to decode the stream - see below. (default: CRLF)
binary: whether the data is binary (true) or text (false). (default: false)
charset: the charset used when converting text to String. (default: UTF-8)

Available Decoders

Text Data

CRLF (default): text terminated by carriage return (0x0d) followed by line feed (0x0a)
LF: text terminated by line feed (0x0a)
NULL: text terminated by a null byte (0x00)
STXETX: text preceded by an STX (0x02) and terminated by an ETX (0x03)

Text and Binary Data

RAW: no structure - the client indicates a complete message by closing the socket
L1: data preceded by a one byte (unsigned) length field (supports up to 255 bytes)
L2: data preceded by a two byte (unsigned) length field (up to 2¹⁶-1 bytes)
L4: data preceded by a four byte (signed) length field (up to 2³¹-1 bytes)

Examples

The following examples all use echo to send data to netcat which sends the data to the source.

The echo options -en allows echo to interpret escape sequences and not send a newline.

CRLF Decoder

$ curl -X POST -d "tcp | file" http://localhost:8080/streams/tcptest

This uses the default (CRLF) decoder and port 1234; send some data

$ echo -en 'foobar\r\n' | netcat localhost 1234

See if the data ended up in the file

$ cat /tmp/xd/output/tcptest

LF Decoder

$ curl -X POST -d "tcp --port=1235 --decoder=LF | file" http://localhost:8080/streams/tcptest2

$ echo -en 'foobar\n' | netcat localhost 1235

$ cat /tmp/xd/output/tcptest2

NULL Decoder

$ curl -X POST -d "tcp --port=1236 --decoder=NULL | file" http://localhost:8080/streams/tcptest3

$ echo -en 'foobar\x00' | netcat localhost 1236

$ cat /tmp/xd/output/tcptest3

STXETX Decoder

$ curl -X POST -d "tcp --port=1237 --decoder=STXETX | file" http://localhost:8080/streams/tcptest4

$ echo -en '\x02foobar\x03' | netcat localhost 1237

$ cat /tmp/xd/output/tcptest4

RAW Decoder

$ curl -X POST -d "tcp --port=1238 --decoder=RAW | file" http://localhost:8080/streams/tcptest5

$ echo -n 'foobar' | netcat localhost 1238

$ cat /tmp/xd/output/tcptest5

L1 Decoder

$ curl -X POST -d "tcp --port=1239 --decoder=L1 | file" http://localhost:8080/streams/tcptest6

$ echo -en '\x06foobar' | netcat localhost 1239

$ cat /tmp/xd/output/tcptest6

L2 Decoder

$ curl -X POST -d "tcp --port=1240 --decoder=L2 | file" http://localhost:8080/streams/tcptest7

$ echo -en '\x00\x06foobar' | netcat localhost 1240

$ cat /tmp/xd/output/tcptest7

L4 Decoder

$ curl -X POST -d "tcp --port=1241 --decoder=L4 | file" http://localhost:8080/streams/tcptest8

$ echo -en '\x00\x00\x00\x06foobar' | netcat localhost 1241

$ cat /tmp/xd/output/tcptest8

Binary Data Example

$ curl -X POST -d "tcp --port=1242 --decoder=L1 | file --binary=true " http://localhost:8080/streams/tcptest9

Note that we configure the file sink with binary=true so that a newline is not appended.

$ echo -en '\x08foo\x00bar\x0b' | netcat localhost 1242

$ hexdump -C /tmp/xd/output/tcptest9
00000000  66 6f 6f 00 62 61 72 0b                           |foo.bar.|
00000008

Home | About | Project | Getting Started | Technical Docs

Contents

Introduction
Getting Started
Running in Distributed Mode
Technical Documentation
Architecture
Streams
Modules
Sources
Processors
Sinks
Taps
Analytics
DSL Reference
Tuples
Samples

Appendix A: Installing Hadoop
Appendix B: Creating a Source Module Appendix C: Creating a Processor Module Appendix D: Creating a Sink Module

Sources

Introduction

HTTP

HTTP with options

Tail

Tail with options

Tail Status Events

Twitter Search

GemFire Continuous Query (CQ)

Launching the XD GemFire Server

Options

Syslog

TCP

TCP with options

Available Decoders

Text Data

Text and Binary Data

Examples

CRLF Decoder

LF Decoder

NULL Decoder

STXETX Decoder

RAW Decoder

L1 Decoder

L2 Decoder

L4 Decoder

Binary Data Example

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally