Streams Configuration

This article explains how to configure stream parameters in meshIQ — covering generic settings, caching, parsing, buffering, and various stream input/output options.

These parameters are applicable to all types of streams.

• UseExecutors  — identifies whether the stream should use an executor service to process activities’ data items asynchronously or not. Default value — false. (Optional)
  • ExecutorThreadsQuantity — defines the executor service thread pool size. Default value — 4. (Optional, applicable only if UseExecutors is set to true)
  • ExecutorQueueDepth — defines the executor service bounded task queue depth. Default value — 2 * ExecutorThreadsQuantity. (Optional, applicable only if UseExecutors is set to true)
  • ExecutorsTerminationTimeout — time to wait (in seconds) for an executor service to terminate. Default value — 20 sec. (Optional, applicable only if UseExecutors is set to true)
  • ExecutorsBoundedModel — identifies whether the executor service should use a bounded task queue model. Default value — false. (Optional, applicable only if UseExecutors is set to true)
    • ExecutorRejectedTaskOfferTimeout — time to wait (in seconds) for a task to be inserted into a bounded queue if the max. queue size is reached. Default value — 20 sec. (Optional, applicable only if ExecutorsBoundedModel is set to true)
  • ExecutorsImmediateShutdown— flag indicating whether the executor service shutdown shall be immediate (dropping all pending tasks) or graceful (trying to complete processing of pending tasks). Default value — false, meaning executors shall shut down gracefully. (Optional)
• PingLogActivityDelay— defines the repetitive interval in seconds between "ping" log entries with stream statistics. Default value — -1 meaning NEVER. (Optional)
• Set of user-defined stream context properties. To define a stream context property, add the ctx: prefix to the property name. These properties are not used directly by the stream itself, but can be used in stream-bound parsers configuration over dynamic locators or variable expressions to enrich parsing context. (Optional)

Sample:

<property name="UseExecutors" value="true"/>
<property name="ExecutorThreadsQuantity" value="5"/>
<property name="ExecutorsTerminationTimeout" value="20"/>
<property name="ExecutorsBoundedModel" value="true"/>
<property name="ExecutorRejectedTaskOfferTimeout" value="20"/>
<property name="ExecutorsImmediateShutdown" value="true"/>
<!-- to define "ping" log entry every 30sec. -->
<property name="PingLogActivityDelay" value="30"/>

• MaxSize — max. capacity of the stream resolved-values cache. Default value — 100. (Optional)

• ExpireDuration — stream resolved-values cache entries’ expiration duration in minutes. Default value — 10. (Optional)

• Persisted — flag indicating cache contents have to be persisted to file on close and loaded on initialization. Default value — false. (Optional)

• FileName — defines file name to persist cache entries as XML. Default value — ./persistedCache.xml. (Optional)

• PersistingPeriod — cache persisting period in seconds. Value <= 0 disables periodic persisting. Default value — 0. (Optional)

Sample:

<property name="MaxSize" value="500"/>
<property name="ExpireDuration" value="30"/>
<property name="Persisted" value="true"/>
<property name="FileName" value="./storage/MyStreamCache.xml"/>
<property name="PersistingPeriod" value="300"/>

These parameters are applicable to streams which use parsers to parse incoming RAW activity data.

• HaltIfNoParser — if set to true, the stream will halt if none of the parsers can parse the activity object RAW data. If set to false — puts a log entry and continues. Default value — false. (Optional, deprecated — use NoParserAction instead)

• NoParserAction — property identifies what the stream shall do if none of the parsers accepts an activity RAW data item. It can be one of: SKIP, HALT, and DUMP. Dump definition like

  DUMP:/var/lib/tnt4j-streams/noparser

  means that all unparsed activity RAW data items shall be stored in /var/lib/tnt4j-streams/noparser directory giving file name as

  [STREAM_NAME]_[TIMESTAMP_IN_MICROSECONDS].dump

  Default value — SKIP. (Optional)

• GroupingActivityName — name of ACTIVITY entity used to group excel workbook streamed events. (Optional)

Sample:

<property name="HaltIfNoParser" value="true"/>
<property name="GroupingActivityName" value="Events from XLSX file"/>

• BufferSize — maximal buffer queue capacity. Default value — 1024. (Optional)

• FullBufferAddPolicy — defines policy for adding a new RAW activity data entry when the buffer queue is full: WAIT or DROP. Default value — WAIT. (Optional)

Sample:

<property name="BufferSize" value="2048"/>
<property name="FullBufferAddPolicy" value="DROP"/>

Stream output can be configured using these configuration properties:

• TNT4JConfigFile — path of tnt4j.properties file. May be used to override default TNT4J system property tnt4j.config defined value. Default value — null. (Optional)

• TNT4JProperty — defines specific TNT4J configuration properties to be used by stream output. (Optional)

• TNT4JConfigZKNode — defines ZooKeeper path where stream configuration is located. Default value — `` (empty). (Optional)

• RetryStateCheck — flag indicating whether tracker state check shall be performed repeatedly, or number of retries to perform. If false, then streaming process exits with java.lang.IllegalStateException on first failure. If true, then state-check retry procedure repeats until success (may repeat an infinite number of times). Default value — true. (Optional)

• RetryPeriod — period in seconds to wait before the next issue of state check or activity recording operation after failure. Default value — 10 sec. (Optional)

• RetryInterval — alias for RetryPeriod. (Optional)

• ResolveServerFromDNS — flag indicating whether to resolve activity entity host name/IP from DNS server. Default value — false. (Optional, deprecated — use parser metadata field @ResolveServerFromDNS@ to set value for individual entities)

• SplitRelatives — flag indicating whether to send activity entity child entities independently, merging data from both parent and child entity fields into the produced entity. Default value — false. (Optional, deprecated — use parser metadata field @SplitRelatives@ to set value for individual entities). NOTE: This value has alias TurnOutActivityChildren left for backward compatibility, but it is not recommended to use it anymore — change it right away for existing configurations.

• BuildSourceFQNFromStreamedData — flag indicating whether to set streamed activity entity Source FQN built from activity fields data instead of the default one configured in tnt4j.properties. Default value — true. (Optional)

• SourceFQN — Source FQN pattern to be used when building it from streamed activity entity fields’ values. Format is:

  SourceType1=${FieldName1}#SourceType2=${FieldName2}#SourceType3=${FieldName3}...

  Default value —

  APPL=${ApplName}#USER=${UserName}#SERVER=${ServerName}#NETADDR=${ServerIp}#GEOADDR=${Location}

  (Optional)

• SendStreamStates — flag indicating whether to send stream status change messages (startup/shutdown) to the output endpoint, e.g., meshIQ. Default value — true. (Optional)

Sample:

<property name="TNT4JConfigFile" value="../../configuration/tnt4j.properties"/>
<tnt4j-properties>
<property name="event.formatter" value="com.jkoolcloud.tnt4j.streams.utils.RedirectTNT4JStreamFormatter"/>
</tnt4j-properties>
<property name="TNT4JConfigZKNode" value="/samples/core/logstash"/>

<property name="RetryStateCheck" value="true"/>
<property name="RetryStateCheck" value="5"/>

<property name="RetryPeriod" value="3"/>
<property name="ResolveServerFromDNS" value="true"/>
<property name="SplitRelatives" value="true"/>
<property name="BuildSourceFQNFromStreamedData" value="false"/>
<property name="SourceFQN" value="APPL=${ApplName}#USER=${UserName}#SERVER=${ServerName}"/>
<property name="SendStreamStates" value="false"/> 

Stream output configuration parameters can be defined under the stream tag (will drill down to the default stream output instance), or under a java-object tag referring to the output type class and referenced from stream like this:

<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="ProgressEventParser">
        <.../>
    </parser>

    <stream name="WmqStream" class="com.jkoolcloud.tnt4j.streams.inputs.WmqStream">
        <property name="SplitRelatives" value="true"/>
        <.../>

        <parser-ref name="ProgressEventParser"/>
    </stream>
</tnt-data-source>

or

<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <java-object name="WmqStreamOutput" class="com.jkoolcloud.tnt4j.streams.outputs.JKCloudActivityOutput">
        <property name="SplitRelatives" value="true"/>
    </java-object>

    <parser name="ProgressEventParser">
        <.../>
    </parser>

    <stream name="WmqStream" class="com.jkoolcloud.tnt4j.streams.inputs.WmqStream">
        <.../>

        <parser-ref name="ProgressEventParser"/>

        <reference name="WmqStreamOutput"/>
    </stream>
</tnt-data-source>

• Attributes:
  • tags — tags list (delimited using ,) for a parser. Tags are used to map the incoming parseable data package with the parser instance dedicated to parse it.

Sample:

<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="ProgressEventParser">
        <.../>
    </parser>

    <stream name="WmqStream" class="com.jkoolcloud.tnt4j.streams.inputs.WmqStream">
        <property name="SplitRelatives" value="true"/>
        <.../>

        <parser-ref name="ProgressEventParser" tags="Kafka,Http"/>
    </stream>
</tnt-data-source>

• Tags:
  • matchExp — defines context or data match criteria (expression) for a parser reference. If the match criteria do not match the incoming data or parsing context, that parser will not be applied to parse the data.

Sample:

<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="ProgressEventParser">
        <.../>
    </parser>

    <stream name="WmqStream" class="com.jkoolcloud.tnt4j.streams.inputs.WmqStream">
        <property name="SplitRelatives" value="true"/>
        <.../>

        <parser-ref name="ProgressEventParser">
            <matchExp>regex:(([a-zA-Z]*):)?(.+)</matchExp>
        </parser-ref>
    </stream>
</tnt-data-source>

It is possible to define context or data match criteria for a parser reference. This allows applying a parser to provided data only when the data or parsing context matches the defined evaluation expression. One parser reference can have multiple match expressions. In that case, all of them must evaluate to a positive match for the parser to be applied.

when a parser reference has no match expressions defined, only an incompatible data type prevents it from being applied.

Sample on field stacked parsers binding with match expressions:

<field name="MessageFormats" locator="MQGACF_ACTIVITY_TRACE.MQBACF_MESSAGE_DATA" locator-type="Label" datatype="String"
       format="string" transparent="true">
    <parser-ref name="BNYM_XML_Msg_Data_Parser" aggregation="Merge">
        <matchExp>xpath:contains(/Request/messageDetails/tag21, 'VOLUME')</matchExp>
    </parser-ref>
    <parser-ref name="Test_7_Mixed_NV_Offsets_Lines_Parser" aggregation="Merge">
        <matchExp>jpath:[?(@.Request.header.receiver == 'SEB')]</matchExp>
        <matchExp>${ObjectName} == "PAYMENTS_QUEUE"</matchExp>
    </parser-ref>
    <parser-ref name="SWIFT_2_Parser" aggregation="Merge">
        <matchExp>contains(:)</matchExp>
    </parser-ref>
</field>

If a match expression definition breaks configuration XML validity, surround it with <![CDATA[ ... ]]>, e.g.:

<parser-ref name="Test_7_Mixed_NV_Offsets_Lines_Parser">
    <matchExp><![CDATA[
        jpath:[?(@.Request.header.receiver == 'SEB')]
    ]]></matchExp>
    <matchExp><![CDATA[
        ${ObjectName} == "PAYMENTS_QUEUE"
    ]]></matchExp>
</parser-ref>

Types of Data Match Evaluation Expressions

• String (default)— it must contain a method name from the StringUtils class and arguments. The method's first parameter is implied and not specified; it's always bound to the value of the message data being parsed. Only subsequent parameters need to be defined in the expression (e.g., contains(PAYMENT), isEmpty()). To negate the expression result, start the expression definition with the ! negate character.

Some samples:

<parser-ref ...>
    <matchExp>isAlpha()</matchExp> <!-- NOTE: default type is applied-->
    <matchExp>string:startsWithIgnoreCase(SWIFT)</matchExp>
    <matchExp>string:!isEmpty()</matchExp>
    <matchExp>string:contains(SWIFT)</matchExp>
</parser-ref>

• RegEx — should contain any valid Regular Expression (RegEx). Positive match is obtained when the RegEx pattern matches some subsequence within the provided data string.

<parser-ref ...>
    <matchExp>regex:(:33A:)(.*)(:24B:)(.*)</matchExp>
    <matchExp>regex:((TID)=(.[^\s:,%]*))|((USER_DATA)=(.[^\s:,%]*))</matchExp>
    <matchExp>regex:(([a-zA-Z]*):)?(.+)</matchExp>
</parser-ref>

• Google RE2 — should contain any valid Google RE2 expression. Positive match is obtained when the Google RE2 pattern matches some subsequence within the provided data string.

<parser-ref ...>
    <matchExp>re2:(:33A:)(.*)(:24B:)(.*)</matchExp>
    <matchExp>re2:((TID)=(.[^\s:,%]*))|((USER_DATA)=(.[^\s:,%]*))</matchExp>
    <matchExp>re2:(([a-zA-Z]*):)?(.+)</matchExp>
</parser-ref>

• JsonPath — should contain any valid JsonPath expression.

• If the expression evaluates as boolean, that value is returned.

• If the expression evaluates as a collection, then to get a positive match it must be not empty.

• In other cases, to get a positive match, the expression should evaluate to a non-null object.

<parser-ref ...>
    <matchExp>jpath:$.connected</matchExp>
    <matchExp>jpath:$.[?(@.id == 'sensor03')]</matchExp>
    <matchExp>jpath:$.[?(@.temperature > 70)]</matchExp>
</parser-ref>

• JMESPath — should contain any valid JMESPath expression.

• If the expression evaluates as boolean, that value is returned.

• If the expression evaluates as a collection, then to get a positive match it must be not empty.

• In other cases, to get a positive match, the expression should evaluate to a non-null object.

<parser-ref ...>
    <matchExp>jmes:connected</matchExp>
    <matchExp>jmes:[?id == 'sensor03']</matchExp>
    <matchExp>jmes:[?temperature > 70]</matchExp>
</parser-ref>

• XPath — should contain any valid XPath expression. If the expression evaluates as boolean, then this value is returned. In other cases, to get a positive match, the expression should evaluate as a non-empty string.

<parser-ref ...>
    <matchExp>xpath:contains(/Request/messageDetails/tag21, 'VOLUME')</matchExp>
    <matchExp>xpath:/Request/genericTransFields/receiver</matchExp>
    <matchExp>xpath:/Request/header[currency='EUR']/messageId</matchExp>
</parser-ref>

• Groovy or JavaScript script-based expressions. To access the field value (data passed to the parser), use the predefined variable placeholder: $fieldValue.

<parser-ref ...>
    <matchExp>groovy:$fieldValue instanceof Map</matchExp>
    <matchExp>javascript:$fieldValue.length > 300</matchExp>
</parser-ref>

Parsing context is treated as parsers’ resolved field values. That is why context expressions should contain variables referencing activity entity field names.

Types of Context Evaluation Expressions

• Groovy (default) — should contain a valid Groovy-language-based boolean expression.

<parser-ref ...>
    <matchExp>${ObjectName}.startsWith("SYSTEM.")</matchExp> <!-- NOTE: default type is applied-->
    <matchExp>groovy:${ObjectName} == "PAYMENTS_QUEUE"</matchExp>
    <matchExp>groovy:${QueueDepth} >= 500</matchExp>
</parser-ref>

• JavaScript — should contain a valid JavaScript-language-based boolean expression.

<parser-ref ...>
    <matchExp>js:${ObjectName}.toUpperCase().indexOf("PAYMENT") == 0</matchExp>
    <matchExp>jscript:${ObjectName} != null</matchExp>
    <matchExp>javascript:${QueueDepth} != 500</matchExp>
</parser-ref>

• XPath — should contain a valid XPath-based boolean expression.

<parser-ref ...>
    <matchExp>xpath:boolean(${ObjectStatus})</matchExp>
    <matchExp>xpath:${EventName} = 'foo'</matchExp>
    <matchExp>xpath:${PaymentSum} >= 500</matchExp>
    <matchExp>xpath:boolean(ts:getFileName(${FilePath}))</matchExp>
</parser-ref>

The last match expression uses our custom XPath function getFileName, wrapped with the XPath boolean() function, to test if a file name was found.

There are many XPath functions that can be used. See the section TNT4J-Streams predefined custom XPath functions and Oracle's Using XPath Functions reference.

• FileName — The system-dependent file name, or a file-name pattern defined using wildcard characters * and ?. (Required)

• FilePolling — Flag true/false indicating whether files should be polled for changes. If not, files are read from oldest to newest sequentially one time. Default: false. (Optional)

  • FileReadDelay — Delay in seconds between file-reading iterations. Applicable only if FilePolling is true. Default: 15 sec. (Optional)

• RestoreState— Flag true/false or file path (if true, the default file is ./<streamName>.TNT4JStreamed.xml) indicating whether file read state should be stored and restored on stream restart. Note: if StartFromLatest is false, read-state storing stays on, but the previously stored read state is reset (no need to delete the state file manually). Default: false. (Optional)

• StartFromLatest — Flag true/false indicating that streaming should start from the latest file entry line. If false, all lines from available files are streamed on startup. null (omitted in configuration) indicates default stream behavior depending on RestoreState:

  • true/filePath — restore saved streaming state
  • false/<empty> — stream from the beginning of the first available file
    Applicable only if FilePolling or RestoreState is set. Default: null. (Optional)

• RangeToStream — Colon-separated range of file line numbers to parse and stream to jKoolCloud. Default: 1:. (Optional)

• ActivityDelim — Activities data delimiter used by the stream. Values: EOL (end of line), EOF (end of file/stream), or any user-defined symbol/string. Default: EOL. (Optional)

• KeepLineSeparators — Flag indicating whether to return line separators at the end of the read line. Default: false. (Optional)

• TruncatedFilePolicy — Truncated file access policy (when the file size decreases while it is streamed). Values: START_FROM_BEGINNING (read from beginning), CONTINUE_FROM_LAST (continue from last line, skipping available lines). Default: START_FROM_BEGINNING. (Optional)

• Charset — Charset name used to decode file data. Must be resolvable by java.nio.charset.Charset#forName(String). Value guess indicates the stream (except HDFS) should guess the charset using the first bytes of the file. Default: the value returned by java.nio.charset.Charset#defaultCharset(). (Optional)

Sample:

<property name="FileName" value="C:/Tomcat_7_0_34/logs/localhost_access_log.*.txt"/>
<property name="FileReadDelay" value="5"/>
<property name="StartFromLatest" value="true"/>
<property name="FilePolling" value="true"/>
<property name="RestoreState" value="true"/>
<property name="RangeToStream" value="12:125"/>
<property name="ActivityDelim" value="-----"/>
<property name="KeepLineSeparators" value="true"/>
<property name="TruncatedFilePolicy" value="CONTINUE_FROM_LAST"/>
<property name="Charset" value="UTF-16LE"/>

When using HDFS, the file name is defined using a URL like hdfs://[host]:[port]/[path]. The path may contain wildcards.

Also see Generic streams parameters and Buffered streams parameters.

Accessible via the $METADATA$ locator:

• MD_FILE_NAME — File name the line was read from

• MD_FILE_CREATE_TS — File creation timestamp

• MD_FILE_ACCESS_TS — File access (read) timestamp

• MD_FILE_MODIFY_TS — File modify (write) timestamp

• MD_LINE_NUMBER — Line number in the file

When the file is rolling, the file name and timestamps may change at roll time.

• FileName — The system-dependent file name. (Required — exactly one of FileName or Port)

• Port — Port number to accept character stream over TCP/IP. (Required — exactly one of FileName or Port)

• RestartOnInputClose — Flag indicating to restart the stream if the input socket gets closed. Default: false. (Optional)

Sample:

<property name="FileName" value="messages.json"/>

or

<property name="Port" value="9595"/>
<property name="RestartOnInputClose" value="true"/>

Also see Generic streams parameters.

There can be only one parser referenced by this stream.

When data is fed from file:

• MD_FILE_NAME — File name the feed is read from

• MD_FILE_CREATE_TS — File creation timestamp

• MD_FILE_ACCESS_TS — File access (read) timestamp

• MD_FILE_MODIFY_TS — File modify (write) timestamp

When data is fed from socket:

• MD_SERVER_PORT — Server port number the stream feed was initiated on

• MD_LOCAL_ADDRESS — Local address the feed is reading data from

• MD_LOCAL_PORT — Local port number the feed is reading data from

• MD_REMOTE_ADDRESS — Remote address activity data is fed from

• MD_REMOTE_PORT — Remote port number activity data is fed from

Supported file systems:

• SCP - schemes ssh.unix:, scp: and ssh:
• SFTP - scheme sftp:
• ZIP - schemes zip: and jar:
• FILE - scheme file:
• Any other JSR-203–compliant provider (requires adding the implementation libraries to the classpath). These are not tested and may not work out of the box.

When accessing files over SCP (ssh.unix:, scp:, ssh:) from *nix machines, ensure the environment variable SSH_AUTH_SOCK is configured.

General Stream Configuration Parameters

• Host — Remote machine host name/IP address. Default: localhost. (Optional — can be defined via FileName, e.g., ftp://username:password@hostname:port/[FILE_PATH])

• Port — Remote machine port number. (Optional — can be defined via FileName)

• UserName — Remote user name. (Optional — can be defined via FileName)

• Password — Remote user password. (Optional — can be defined via FileName)

• Scheme — File access protocol scheme. Default: scp. (Optional — can be defined via FileName, e.g., ssh.unix:///[FILE_PATH])

• StrictHostKeyChecking — Whether to strictly check remote host key changes against SSH known hosts:

  • If flag is set to yes, SSH will never automatically add host keys to the ~/.ssh/known_hosts file. It will also refuse to connect to a host if its host key has changed. While this provides maximum protection against trojan horse attacks, it can be troublesome when the /etc/ssh/ssh_known_hosts file is poorly maintained or when connections to new hosts are frequent. In such cases, the user must manually add all new hosts.
  • If this flag is set to no, If this flag is set to no, SSH will automatically add new host keys to the user’s known hosts files.
    The host keys of known hosts are always verified, regardless of this setting. The flag value must be either yes or no. Default value - no. (Optional)

• IdentityFromPrivateKey — Private key file for secure connection initialization. (Optional)

• KnownHosts — Known hosts file path. (Optional)

• ResolveAbsolutePath — Whether to resolve an absolute file path when a relative one is provided. Default: false. (Optional)

• Plus the set of target JSR-203 FileSystemProvider-supported properties.

Sample:

<property name="Host" value="172.16.6.26"/>
<property name="Port" value="22"/>
<property name="UserName" value="osboxes"/>
<property name="Password" value="slabs"/>
<property name="Scheme" value="sftp"/>
<property name="StrictHostKeyChecking" value="no"/>
<property name="KnownHosts" value="/home/joe/.ssh/known_hosts"/>
<property name="IdentityFromPrivateKey" value="/home/joe/.ssh/id_dsa"/>
<property name="ResolveAbsolutePath" value="true"/>

Sample:

<property name="FileName" value="ssh.unix:///home/osboxes/single-log/orders.log"/>
<property name="RestartOnInputClose" value="true"/>

Also see JSR-203 FileSystem streams parameters and Generic streams parameters.

Sample:

<property name="FileName" value="ssh.unix:///home/osboxes/single-log/orders.log"/>

Also see JSR-203 FileSystem streams parameters and File line stream parameters (also applies for HDFS).

• InputCloseable – flag indicating whether the stream should close the input when the stream is closing. Default value – true. (Optional)
• Charset – charset name used to decode the data contained in the file(s). The charset name must comply with the Java specification (be resolvable by java.nio.charset.Charset#forName(String)) to be handled properly. Default value – the one returned by java.nio.charset.Charset#defaultCharset(). (Optional)

Sample:

<property name="InputCloseable" value="false"/>
<property name="Charset" value="UTF-16LE"/>

Also see Generic streams parameters and Parseable streams parameters.

This stream does not have any additional configuration parameters.

Also see Standard Java input stream parameters.

Stream class: com.jkoolcloud.tnt4j.streams.inputs.SocketInputStream

• Port — Port number to run the server socket. Default: 12569. (Optional)

Also see Generic streams parameters and Buffered streams parameters.

Stream class: com.jkoolcloud.tnt4j.streams.inputs.HttpStream

• Host — Host name/IP to run the HTTP server. Defaults: localhost, 127.0.0.1, and the canonical hostname of the runner machine. (Optional)

• Port — Port number to run the HTTP server. Default: 8080. (Optional)

• UseSSL — Flag indicating whether to use SSL. Default: false. (Optional)
  • Keystore — Keystore path. (Optional; applicable only if UseSSL is true)
  • KeystorePass — Keystore password. (Optional; applicable only if UseSSL is true)
    • KeyPass — Key password. (Optional; applicable only if UseSSL is true)

Sample:

<property name="Port" value="8081"/>
<property name="UseSSL" value="true"/>
<property name="Keystore" value="path_to_keystore_file"/>
<property name="KeystorePass" value="somePassword"/>
<property name="KeyPass" value="somePassword"/>

Also see Generic streams parameters and Buffered streams parameters.

• java.naming.provider.url — JMS server URL. (Required)

• Queue — Queue destination name (or names delimited by ,). (Required — at least one of Queue or Topic)

• Topic — Topic destination name (or names delimited by ,). (Required — at least one of Queue or Topic)

• java.naming.factory.initial — JNDI context factory name. (Required)

• JMSConnFactory — JMS connection factory name. (Required)

• Supports the set of JNDI context configuration properties provided by the JMS server implementation. See javax.naming.Context for details. (Optional)

Sample:

<property name="java.naming.provider.url" value="tcp://localhost:61616"/>
<property name="Topic" value="topic.SampleJMSTopic,topic.OtherSampleJMSTopic"/>
<property name="java.naming.factory.initial" value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
<property name="JMSConnFactory" value="ConnectionFactory"/>
<parser-ref name="SampleJMSParser"/>

or

<property name="java.naming.provider.url" value="tcp://localhost:61616"/>
<property name="Queue" value="queue.SampleJMSQueue,queue.OtherSampleJMSQueue"/>
<property name="java.naming.factory.initial" value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
<property name="JMSConnFactory" value="ConnectionFactory"/>
<parser-ref name="SampleJMSParser"/>

Also see Generic streams parameters and Buffered streams parameters.

• Topic — Set of topic names (delimited by |) to listen to. (Required — exactly one of Topic or TopicPattern)

• TopicPattern — Topic name RegEx pattern. (Required — exactly one of Topic or TopicPattern)

• Offset — List of topic offsets (delimited by |) from which to start consuming messages. Value -1 or || (for tokenized definitions) means use latest topic offset. The number of offset tokens must match the number of defined topics; a single value applies to all topics. Default: -1 (from latest). (Optional)

• PollTimeoutMs — Messages polling timeout in milliseconds. -1 means no timeout. Default: 1000 (1 sec). (Optional)

• TopicBindingStrategy — How the topic is bound to the consumer instance: SUBSCRIBE or ASSIGN. Default: SUBSCRIBE. (Optional)

• PartitionsUpdatePeriodSec — Period (in seconds) for scheduling updates and assigning active topic partitions for the consumer instance. Default: 60. (Optional; applicable only if TopicBindingStrategy is ASSIGN)

• FileName — Kafka consumer configuration file (consumer.properties) path. (Optional)

• Supports the full set of Kafka Consumer configuration properties. See Kafka Consumer configuration reference.

Sample:

<property name="Topic" value="TNT4JKafkaTestTopic"/>
<property name="Offset" value="2"/>

<property name="Topic" value="TNT4JKafkaTestTopic|TNT4JKafkaProductionTopic"/>
<property name="Offset" value="2|0"/>

<property name="Topic" value="TNT4JKafkaTestTopic|TNT4JKafkaProductionTopic"/>
<property name="Offset" value="2|-1"/>

<property name="Topic" value="TNT4JKafkaTestTopic|TNT4JKafkaProductionTopic|TNT4JKafkaRnDTopic"/>
<property name="Offset" value="2||12"/>

<property name="Topic" value="TNT4JKafkaTestTopic|TNT4JKafkaProductionTopic|TNT4JKafkaRnDTopic"/>
<property name="Offset" value="5"/>

<property name="TopicPattern" value="tnt4j-.*"/>

<property name="PollTimeoutMs" value="3000"/>
<property name="TopicBindingStrategy" value="ASSIGN"/>
<property name="PartitionsUpdatePeriodSec" value="120"/>

<property name="FileName" value="./config/consumer.properties"/>

<property name="bootstrap.servers" value="localhost:6667"/>
<property name="group.id" value="tnt4j-streams-kafka"/>
<property name="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
<property name="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
<property name="enable.auto.commit" value="true"/>
<property name="auto.commit.interval.ms" value="1000"/>
<property name="session.timeout.ms" value="30000"/>
<property name="client.id" value="tnt4j-streams-kafka-consumer-stream"/>

Also see Generic streams parameters and Buffered streams parameters.

• ServerURI — MQTT server URI. (Required)

• TopicString — The topic to subscribe to, which can include wildcards. (Required)

• UserName — Authentication username. (Optional)

• Password — User password. (Optional)

• UseSSL — Flag indicating to use SSL. Default: false. (Optional)

• Keystore — Keystore path. (Optional; applicable only if UseSSL is set to true)

• KeystorePass — Keystore password. (Optional; applicable only if UseSSL is set to true)

Sample:

<property name="ServerURI" value="tcp://localhost:1883"/>
<property name="Topic" value="TNT4JStreams"/>
<property name="UserName" value="someUser"/>
<property name="Password" value="somePassword"/>
<property name="UseSSL" value="true"/>
<property name="Keystore" value="path_to_keystore_file"/>
<property name="KeystorePass" value="somePassword"/>

Also see Generic streams parameters and Buffered streams parameters.

• QueueManager — Queue manager name. (Optional)

• Queue — Queue name. (Required — at least one of Queue, Topic, Subscription, TopicString)

• Topic — Topic name. (Required — at least one of Queue, Topic, Subscription, TopicString)

• Subscription — Subscription name. (Required — at least one of Queue, Topic, Subscription, TopicString)

• TopicString — Topic string. (Required — at least one of Queue, Topic, Subscription, TopicString)

• Host — WMQ connection host name. Also supports WMQ connection formats HOST(PORT) and HOST:PORT. Can have multiple values delimited by ,. Alias for CMQC.HOST_NAME_PROPERTY Queue Manager connection property. (Optional)

• Port — WMQ connection port number. Alias for CMQC.PORT_PROPERTY Queue Manager connection property. Default: 1414. (Optional)

• UserName — WMQ user identifier. Alias for CMQC.USER_ID_PROPERTY Queue Manager connection property. (Optional)

• Password — WMQ user password. Alias for CMQC.PASSWORD_PROPERTY Queue Manager connection property. (Optional)

• Channel — Server connection channel name. Alias for CMQC.CHANNEL_PROPERTY Queue Manager connection property. Default: SYSTEM.DEF.SVRCONN. (Optional)

• StripHeaders — Identifies whether the stream should strip WMQ message headers. Default: true. (Optional)

• MsgReadTimeoutMs — Message GET timeout in milliseconds. -1 means no timeout applied. Default: 10sec. (Optional)

• StreamReconnectDelay — Delay in seconds between queue manager reconnection or failed queue GET iterations. Default: 15sec. (Optional)

• OpenOptions — Defines the open options value used to access a queue or topic. You can define a numeric options value or a concatenation of MQ constant names/values delimited by |. If the options definition starts with !, it means this options set should be used as complete and passed to the Queue Manager without changes. By default, these open options are appended to the predefined sets:

• Predefined set of open options for queue:

• MQOO_FAIL_IF_QUIESCING

• MQOO_INPUT_AS_Q_DEF

• MQOO_SAVE_ALL_CONTEXT

• MQOO_INQUIRE

• Predefined set of open options for topic:

• MQSO_FAIL_IF_QUIESCING

• MQSO_CREATE

• MQSO_MANAGED — if subscription name is empty

• MQSO_RESUME — if subscription name is defined

• (Optional)

• CMQC.XXXXXXX_PROPERTY — Any CMQC-defined Queue Manager connection property. You can define multiple CMQC connection properties per stream definition, but only one per property definition. (Optional)

Sample:

<property name="QueueManager" value="QMGR"/>
<property name="Queue" value="SYSTEM.ADMIN.TRACE.ACTIVITY.QUEUE"/>
<property name="Host" value="wmq.sample.com"/>
<property name="Port" value="1420"/>
<property name="UserName" value="Administrator"/>
<property name="Password" value="someUserPass"/>
<property name="Channel" value="SYSTEM.DEF.SVRCONN2"/>
<property name="StripHeaders" value="false"/>
<property name="MsgReadTimeoutMs" value="3000"/>
<property name="StreamReconnectDelay" value="30"/>
<property name="OpenOptions" value="!MQSO_FAIL_IF_QUIESCING|MQSO_CREATE|MQSO_MANAGED|MQSO_WILDCARD_CHAR"/>
<property name="CMQC.USE_MQCSP_AUTHENTICATION_PROPERTY" value="true"/>
<property name="CMQC.THREAD_AFFINITY_PROPERTY" value="false"/>
<.../>
<!-- MULTIPLE INSTANCES (MI) host configurations -->
<property name="Host" value="wmq.sample1.com(1420),wmq.sample2.com(1421)"/>
<property name="Host" value="wmq.sample1.com:1420,wmq.sample2.com:1421"/>
<property name="Host" value="wmq.sample1.com,wmq.sample2.com"/>
<property name="Port" value="1420"/>
<.../>

Also see Generic streams parameters.

• TraceOperations — Defines the traced MQ operations name filter mask (wildcard or RegEx) to process only traces of MQ operations whose names match this mask. Default: *. (Optional)

• ExcludedRC — Defines a set of excluded MQ trace events reason codes (delimited using |) to process only MQ trace events having reason codes not contained in this set. Set entries may be defined using both numeric and MQ constant name values. Default: ````. (Optional)

• SuppressBrowseGets — Flag indicating whether to exclude WMQ BROWSE-type GET operation traces from streaming. Default: false. (Optional)

Sample:

<property name="TraceOperations" value="MQXF_(GET|PUT|CLOSE)"/>
<property name="ExcludedRC" value="MQRC_NO_MSG_AVAILABLE|30737"/>
<property name="SuppressBrowseGets" value="true"/>

Also see WMQ Stream parameters.

See post on how to configure WMQ server and client sides communicating over SSL.

When you already have server-side SSL configured and SSL certificates imported into the client machine (if it is not the same as the server) KeyStore/TrustStore, to run the WMQ stream you’ll need:

• In streams configuration, define Cipher Suite used for the SSL connection:

<property name="CMQC.SSL_CIPHER_SUITE_PROPERTY" value="TLS_RSA_WITH_AES_128_CBC_SHA256"/>

The cipher suite value must match the one configured for the Server Connections Channel!

• To run the stream instance using SSL communication, define Java KeyStore/TrustStore credentials via system properties:

  -Djavax.net.ssl.trustStore=[sysPath]/[ts_file].jks
  -Djavax.net.ssl.trustStorePassword=clientpass
  -Djavax.net.ssl.keyStore=[sysPath]/[ks_file].jks
  -Djavax.net.ssl.keyStorePassword=clientpass

• If you are using a non-IBM JVM to run streams (WMQ client),

  set:-Dcom.ibm.mq.cfg.useIBMCipherMappings=false

  to disable direct cipher suites mapping between the client and server sides. 

• Enable verbose JVM debug messages for SSL communication (optional):

   data-start="5052" data-end="5123"-Djavax.net.debug=ssl

Sample:

<property name="FileName" value="./tnt4j-streams-core/samples/zip-stream/sample.gz"/>
<property name="ArchType" value="GZIP"/>
<property name="Charset" value="UTF-16LE"/>

In case of using Hdfs, file name is defined using URL like hdfs://[host]:[port]/[path]. Zip entry name may contain wildcards.

Also see Generic streams parameters.

All WS streams have a special configuration section, called a scenario. Streaming scenarios allow for defining steps. A step defines request/invocation/execution parameters and scheduler. Steps are invoked/executed independently of each other.

scenario tag

• Required attribute: name (any string)

• Optional attributes:
  url (service request URL), username (service authentication user), password (service authentication password), and method (GET/POST — default value GET).
  These service access attributes are applied for all scenario steps if the scenario does not define individual values of attributes.

  • property tag has attributes:
    • name — scenario configuration property name
    • value — scenario configuration property value. It also can be defined as tag body.

step tag

• Required attribute: name (any string)

• Optional attributes:
  url (service request URL), username (service authentication user), password (service authentication password), and method (GET/POST — default value GET).
  Attributes url, username, password, and method override values defined for corresponding scenario tag attributes. That way some scenario steps can have individual service access configurations.

  • property tag has attributes:
    • name — scenario step configuration property name
    • value — scenario step configuration property value. It also can be defined as tag body.

schedule-cron tag

• Attributes:
  expression — Cron expression. (Required)
  startDelay — Schedule start delay interval as positive integer (zero valid) numeric value. Default: 0. (Optional)
  startDelayUnits — Schedule start delay time units name. Default: SECONDS. (Optional)

schedule-simple tag

• Required attributes:
  interval — Schedule interval as positive integer (non-zero) numeric value. (Required)
  units — Schedule interval time units name. Default: MILLISECONDS. (Optional)

• Optional attributes:
  repeatCount — Schedule repeats count as integer numeric value, where -1 means endless. Default: -1.
  startDelay — Schedule start delay interval as positive integer (zero valid) numeric value. Default: 0.
  startDelayUnits — Schedule start delay time units name. Default: SECONDS.

request tag

request is an XML tag to define string-represented request data (e.g., system command with parameters). To define XML contents, it is recommended to use <![CDATA[]]>. It has an optional attribute id to define a request identifier string used to identify the request when, i.e., example formatting log messages.

parser-ref tag

Used to map received response data and the parser to parse it.

• Attributes:
  name — referenced parser name. (Required)
  tags — parser tags set, used to map parser with parser activity data by some particular values. (Optional)

• Child tags:
  matchExp — Used to define an evaluation expression to check if input data or parsing context matches it and the referenced parser shall parse provided data.

req-param tag

Used to define a request parameter.

• Attributes:
  id — parameter identifier. (Optional)
  value — parameter value. (Required)
  type — parameter value type. (Optional)
  format — parameter value format. (Optional)
  timezone — parameter date-time value format timezone. (Optional)
  transient — Flag indicating whether to add (when value false) the parameter to the service request or just keep it to be used for internal aggregations (when value true), e.g., request/response mapping. Default: false. (Optional)

condition tag

Used to define a request state condition.

• Attributes:
  id — condition identifier. (Optional)
  resolution — condition resolution. (Required) Can be one of enumeration values:
  SKIP — skip request execution when condition matches expression
  STOP — stop and exit stream when condition matches expression

• Child tags:
  matchExp — Used to define an evaluation match expression to check if request context (request parameters value, stream properties, cache values) matches it

Sample:

<!-- Sample scenario for RESTful services request -->
<scenario name="Sample REST stream scenario">
    <step name="Step Kaunas"
          url="http://api.openweathermap.org/data/2.5/weather?q=Kaunas&amp;APPID=<YOUR_APP_TOKEN>&amp;units=metric"
          method="GET">
        <schedule-cron expression="0/15 * * * * ? *"/>
    </step>
    <.../>
    <step name="Step Klaipeda"
          url="http://api.openweathermap.org/data/2.5/weather?q=Klaipeda&amp;APPID=<YOUR_APP_TOKEN>&amp;units=metric"
          method="GET">
        <schedule-simple interval="45" units="Seconds" repeatCount="10"/>
    </step>
</scenario>
<!-- Sample scenario for SOAP services request -->
<scenario name="Sample WS stream scenario">
    <step name="Step 1"
          url="http://wsdot.wa.gov/traffic/api/WeatherInformation/WeatherInformation.svc">
        <schedule-simple interval="35" units="Seconds" repeatCount="10"/>
        <request>
            <![CDATA[
                    SOAPAction:http://tempuri.org/IWeatherInformation/GetCurrentWeatherInformationByStationID
                    <tem:GetCurrentWeatherInformationByStationID xmlns:tem="http://tempuri.org/">
                        <tem:AccessCode>aeb652b7-f6f5-49e6-9bdb-e2b737ebd507</tem:AccessCode>
                        <tem:StationID>1909</tem:StationID>
                    </tem:GetCurrentWeatherInformationByStationID>
                ]]>
        </request>
    </step>
</scenario>
<!-- Sample scenario for System Cmd invocation -->
<scenario name="Sample CMD stream scenario">
    <step name="Step Windows">
        <request>typeperf "\Processor(_Total)\% Processor Time" -sc 1</request>
        <schedule-simple interval="25" units="Seconds" repeatCount="-1"/>
    </step>
</scenario>
<!-- Sample of request and parser mapping -->
<scenario name="RabbitMQ Sampling scenario">
    <step name="All RabbitMQ metrics">
        <schedule-simple interval="30" units="Seconds" repeatCount="-1"/>

        <request>
            python rabbitmqadmin -f raw_json list users

            <parser-ref name="UsersRespParser"/>
        </request>
        <request>
            python rabbitmqadmin -f raw_json list vhosts

            <parser-ref name="HostsRespParser"/>
        </request>
        <request>
            python rabbitmqadmin -f raw_json list overview

            <parser-ref name="OverviewRespParser"/>
        </request>
    </step>
</scenario>
<!-- Sample scenario for JDBC query invocation -->
<scenario name="Sample DB2-JDBC stream scenario">
    <step name="Step Query1" url="jdbc:db2://[HOST]:50000/SB2BIDB" username="[USER_NAME]" password="[USER_PASS]">
        <schedule-simple interval="60" units="Seconds" repeatCount="-1" startDelay="3" startDelayUnits="Minutes"/>

        <request>
            <![CDATA[
                SELECT *
                FROM TRANS_DATA
                WHERE CREATION_DATE > ?
                ORDER by CREATION_DATE DESC
                FETCH FIRST 100 ROWS ONLY
            ]]>
            <req-param id="1" value="${LastRecordCDate}" type="TIMESTAMP"/>

            <condition id="When to stop stream" resolution="STOP">
                <matchExp><![CDATA[
                    {LastRecordIndex} > 2000
                ]]></matchExp>
            </condition>

            <parser-ref name="SampleResultSetParser"/>
        </request>
    </step>
</scenario>

Generic Configuration Properties

• SynchronizeRequests — Flag indicating that stream-issued requests shall be synchronized and handled in configuration-defined sequence — waiting for prior request to complete before issuing the next one. This property can be defined under stream or scenario step tags and synchronizes requests under the according scope. Default: true. (Optional)

• DropRecurrentRequests — Flag indicating whether to drop streaming stream input buffer contained recurring requests when the stream input scheduler invokes requests faster than they can be processed (parsed and sent to sink, e.g., because of sink/JKool limiter throttling). Default: true. (Optional)

• List of Quartz configuration properties. See Quartz Configuration Reference for details. (Optional)

Sample:

<property name="SynchronizeRequests" value="true"/>
<property name="DropRecurrentRequests" value="false"/>
<!-- Quartz configuration -->
<property name="org.quartz.scheduler.instanceName" value="MyStreamScheduler"/>
<property name="org.quartz.threadPool.threadCount" value="5"/>
<.../>

All request parameters and context properties values can be accessed as metadata entries over locator $METADATA$.[PARAM_ID], where [PARAM_ID] is the request parameter and context property identifier.

• DisableSSL — Flag indicating that stream should disable SSL context verification. Default: false. (Optional)

• List of custom WS Stream requests configuration properties. Put a variable placeholder in request/step configuration (e.g., ${WsEndpoint}) and put a property with the same name into the stream properties list (e.g., <property name="WsEndpoint" value="https://192.168.3.3/ws"/>) to have the value mapped into request data. (Optional)

Sample:

<property name="DisableSSL" value="true"/>
<!-- Custom WS request properties -->
<property name="WsEndpoint" value="https://192.168.3.3/ws"/>
<.../>

• SecurityCachedTokenKey — Defines streams cache entry key referring login request received session ID token. Default: Token. (Optional)

• SecurityResponseParserTag — Defines tag value used to map login request data and parser used to parse it. Default: login. (Optional)

Sample:

<property name="SecurityCachedTokenKey" value="SessionIDToken"/>
<property name="SecurityResponseParserTag" value="loginData"/>

Also see Generic streams parameters and Buffered streams parameters.

• MaxTotalPoolConnections — Defines the maximum number of total open connections in the HTTP connections pool. Default: 5. (Optional)

• DefaultMaxPerRouteConnections — Defines the maximum number of concurrent connections per HTTP route. Default: 2. (Optional)

Sample:

<property name="MaxTotalPoolConnections" value="10"/>
<property name="DefaultMaxPerRouteConnections" value="4"/>

This stream request parameters req-param having id prefixed H: will add as HTTP message (GET/POST) headers, e.g.:

<req-param id="H:Authorization" value="Basic dXNlcjpwYXNz"/>
<req-param id="H:Cache-Control" value="no-cache"/>
<req-param id="H:Content-MD5" value="Q2hlY2sgSW50ZWdyaXR5IQ=="/> 

Also see Generic streams parameters and Buffered streams parameters.

<property name="QueryFetchRows" value="500"/>
<property name="QueryMaxRows" value="50000"/>
<!-- HikariCP configuration properties -->
<property name="autoCommit" value="false"/>
<property name="readOnly" value="true"/>
<property name="maximumPoolSize" value="5"/>
<property name="maxLifetime" value="1680000"/> <!-- 28 minutes -->
<property name="idleTimeout" value="600000"/> <!-- 10 minutes -->
<property name="connectionTimeout" value="30000"/> <!-- 30 seconds -->
<!-- Oracle JDBC data source configuration properties -->
<property name="jdbc.v$session.program" value="B2Bi-JDBC-Stream"/>

<property name="FileName" value="tnt4j-stream-activities.log"/>

<property name="Port" value="9009"/>
<property name="RestartOnInputClose" value="true"/>
<property name="BufferSize" value="2048"/>
<property name="FullBufferAddPolicy" value="DROP"/>

<property name="FileName" value="./tnt4j-streams-msoffice/samples/xlsx-rows/sample.xlsx"/>
<property name="SheetsToProcess" value="Sheet(1|8|12)"/>
<property name="WorkbookPassword" value="xlsPass"/>

<property name="RangeToStream" value="5:30"/>