-
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
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
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 one option
-
query: The query that will be run against Twitter
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 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)
-
gemfireHost: The host on which the GemFire server is running. Default is 'localhost'
-
gemfirePort: The port on which the GemFire server is running. Default is 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 each take one option port, which determines 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 (default1234) -
reverse-lookup- perform a reverse DNS lookup on the remote IP Address (defaultfalse) -
socket-timeout- the timeout (ms) before closing the socket when no data received (default120000) -
nio- whether or not to use NIO (default false). NIO is more efficient when there are many connections. -
decoder- how to decode the stream (defaultCRLF) - see below -
binary- whether the data is binary (true) or text (false) - (default false) -
charset- the charset used when converting text toString(default 'UTF-8')
-
CRLF- text terminated by carriage return (0x0d) followed by line feed (0x0a) (TELNET) -
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 2**16-1 bytes) -
L4- data preceded by a four byte (signed) length field (up to 2**31-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
Home | About | Project | Getting Started | Technical Docs