Processors - garyrussell/spring-xd GitHub Wiki

The simplest way to use the filter processor is to pass a SpEL expression when creating the stream. The expression should evaluate the message and return true or false. For example:

This filter will only pass Messages to the log sink if the payload is the word "good". Try sending "good" to the HTTP endpoint and you should see it in the XD log:

Alternatively, if you send the word "bad" (or anything else), you shouldn’t see the log entry.

As part of the SpEL expression you can make use of the pre-registered JSON Path function.

This filter example shows to pass messages to the output channel if they contain a specific JSON field matching a specific value.

Note: There is no space between payload JSON and the jsonPath in the expression

This filter will only pass Messages to the log sink if the JSON payload contains the firstName "John". Try sending this payload to the HTTP endpoint and you should see it in the XD log:

Alternatively, if you send a different firstName, you shouldn’t see the log entry.

Here is another example usage of filter

This is an example that is operating on a JSON payload of tweets as consumed from the twitter search module.

For more complex filtering, you can pass the location of a Groovy script using the script option. If you want to pass variable values to your script, you can statically bind values using the variables option or optionally pass the path to a properties file containing the bindings using the propertiesLocation option.All properties in the file will be made available to the script as variables. Note that payload and headers are implicitly bound to give you access to the data contained in a message.

Example:

By default, Spring XD will search the classpath for custom-filter.groovy and custom-filter.properties. You can place the script in ${xd.home}/modules/processor/scripts and the properties file in ${xd.home}/config to make them available on the classpath. Alternatively, you can prefix the script and properties-location values with file: to load from the file system.

In the following stream definitions, the filter will pass only the first message:

In the following stream definitions, the filter will pass all messages (provided the payload type supports a size() method):

Note the last example demonstrates that values specified in variables override values from propertiesLocation

The simplest way to use the transform processor is to pass a SpEL expression when creating the stream. The expression should return the modified message or payload. For example:

This transform will convert all message payloads to the word "FOO". Try sending something to the HTTP endpoint and you should see "FOO" in the XD log:

As part of the SpEL expression you can make use of the pre-registered JSON Path function. The syntax is #jsonPath(payload,'<json path expression>')

For more complex transformations, you can pass the location of a Groovy script using the script option. If you want to pass variable values to your script, you can statically bind values using the variables option or optionally pass the path to a properties file containing the bindings using the propertiesLocation option. All properties in the file will be made available to the script as variables. Note that payload and headers are implicitly bound to give you access to the data contained in a message. See the Filter example for a more detailed discussion of script variables.

By default, Spring XD will search the classpath for custom-transform.groovy and custom-transform.properties. You can place the script in ${xd.home}/modules/processor/scripts and the properties file in ${xd.home}/config to make them available on the classpath. Alternatively, you can prefix the script and properties-location values with file: to load from the file system.

xd:> stream create --name groovyprocessortest --definition "http --port=9006 | script --script=custom-processor.groovy --variables='x=foo' | log" --deploy

xd:> stream create --name groovyprocessortest --definition "http --port=9006 | script --script=custom-processor.groovy --propertiesLocation=custom-processor.properties | log" --deploy

This splitter converts a JSON message payload to the value of a specific JSON field.

Try sending this payload to the HTTP endpoint and you should see just the value "John" in the XD log:

xd:> stream create --name aggregates --definition "http | aggregator --count=3 --aggregation=T(org.springframework.util.StringUtils).collectionToDelimitedString(#this.![payload],' ') | log" --deploy

xd:> http post --data Hello
xd:> http post --data World
xd:> http post --data !

xd:>stream create earthquakes --definition "trigger | http-client --url='''http://earthquake.usgs.gov/earthquakes/feed/geojson/all/day''' --httpMethod=GET | log" --deploy

The shell processor forks an external process by running a shell command to launch a process written in any language. The process should implement a continual loop that waits for input from stdin and writes a result to stdout in a request-response manner. The process will be destroyed when the stream is undeployed. For example, it is possible to invoke a Python script within a stream in this manner. Since the shell processor relies on low-level stream processing there are some additional requirements:

• Input and output data are expected to be Strings, the charset is configurable.

• The shell process must not write out of band data to stdout, such as a start up message or prompt.

• Anything written to stderr will be logged as an ERROR in Spring XD but will not terminate the stream.

• Responses written to stdout must be terminated using the configured encoder (CRLF or "\r\n" is the default) for the module and must not exceed the configured bufferSize

• Any external software required to run the script must be installed on the container node to which the module is deployed.

Here is a simple Python example that echos the input:

To try this example, copy the above script and save it to echo.py. Start Spring XD and create a stream:

you should see the time echoed in the log:

This script can be easily modified to do some actual work by providing a function that takes the input as an argument and returns a string. Then insert the function call:

The shell processor has the following options:

bufferSize

the size of the buffer (bytes) to use when encoding/decoding (int, default: 2048)

charset

the charset used when converting from String to bytes (String, default: UTF-8)

command

the shell command (String, no default)

encoder

the encoder to use when sending messages (Encoding, default: CRLF, possible values: CRLF,LF,NULL,STXETX,RAW,L1,L2,L4)

environment

additional process environment variables as comma delimited name-value pairs (String, no default)

redirectErrorStream

redirects stderr to stdout (boolean, default: false)

workingDir

the process working directory (String, no default)