Skip to content

Sources

Jump to bottom Edit New page
Gary Russell edited this page May 28, 2013 · 39 revisions

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

HTTP

The HTTP source has X options

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

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

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

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)

  • 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}]

Clone this wiki locally