Sources

Sources

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

The Sources covered are

• HTTP

• Tail

• Gemfire CQ

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

To create a stream definition post using curl

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

Make sure the default output directory exists

$ mkdir -p /tmp/xd/input

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]

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:

$ curl -X POST -d "gemfire-cq --query=Select * from /Stocks where symbol='VMW' | file" http://localhost:8080/streams/cqtest

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=78, symbol=VMW}]
CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=UPDATE; key=VMW; value=PDX[1,__GEMFIRE_JSON]{price=79, symbol=VMW}]
CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=UPDATE; key=VMW; value=PDX[2,__GEMFIRE_JSON]{price=80, symbol=VMW}]