-
Notifications
You must be signed in to change notification settings - Fork 0
Sources
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
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.
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
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
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
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
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]
The twittersearch source has three required parameters
- query
-
The query that will be run against Twitter (required)
- consumerKey
-
An application consumer key issued by twitter
- consumerSecret
-
The secret corresponding to the
consumerKey
To get a consumerKey and consumerSecret you need to register a twitter application. If you don’t already have one set up, you can create an app at the Twitter Developers site to get these credentials.
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
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.
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)
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}]
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.
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
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)
- 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)
- 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 216-1 bytes)
- L4
-
data preceded by a four byte (signed) length field (up to 231-1 bytes)
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.
$ 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
$ 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
$ 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
$ 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
$ 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
$ 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
$ 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
$ 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
$ 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