Add new flow control agents: timer-source and trigger-event (#110)

Co-authored-by: Mendon Kissling <[email protected]>

Author: eolivelli

SUMMARY.md

@@ -100,6 +100,8 @@
     * [unwrap-key-value](pipeline-agents/data-transform/unwrap-key-value.md)
   * [Flow control](pipeline-agents/flow-control/README.md)
     * [dispatch](pipeline-agents/flow-control/dispatch.md)
+    * [timer-source](pipeline-agents/flow-control/timer-source.md)
+    * [trigger-event](pipeline-agents/flow-control/trigger-event.md)
 * [Custom Agents](pipeline-agents/custom-agents/README.md)
   * [Agent Developer Guide](pipeline-agents/agent-developer-guide/README.md)
     * [Agent Types](pipeline-agents/agent-developer-guide/agent-types.md)

building-applications/expression-language.md

@@ -49,6 +49,8 @@ Utility methods available under the `fn` namespace. For example, to get the curr
 | fromJson                          | Parse input as JSON.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
 | lowercase(input)                  | Changes the capitalization of a string. If input is not a string, it attempts a string conversion. If the input is null, it returns null.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | now()                             | Returns the current epoch millis.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
+| uuid()                             | Returns a new UUID (unique id) as string |
+| random(number)                     | Returns a random number between zero the the max value provided as argument |
 | replace(input,regex,replacement)  | Replaces each substring of `input` that matches the `regex` regular expression with `replacement`. See [Java's replaceAll](https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/lang/String.html#replaceAll\(java.lang.String,java.lang.String\)).                                                                                                                                                                                                                                                                                                                                                                                                                |
 | str(input)                        | Converts `input` to a string.|
 | toString(input)                   | Converts `input` to a string.|

pipeline-agents/flow-control/README.md

@@ -15,4 +15,6 @@ layout:
 # Flow control agents
 
 
-* [Dispatch](dispatch.md)
+* [Dispatch](dispatch.md)
+* [Timer source](timer-source.md)
+* [Trigger Event](trigger-event.md)

pipeline-agents/flow-control/dispatch.md

@@ -4,7 +4,7 @@ This agent enables you to dispatch records to different topics, based on conditi
 
 ### Example
 
-Example of reading from a container, extracting the document text, splitting the text into chunks, and outputting each chunk as a message to the output topic.
+This is an example about how to use the `dispatch` agent to dispatch records to different topics based on the language of the text.
 
 ```yaml
   - name: "Detect language"
@@ -26,14 +26,13 @@ Example of reading from a container, extracting the document text, splitting the
           action: drop
 ```
 
-This example detect the language of the input record and then decides to send english and french records to different topics. 
-You could then implement another pipeline to perform different computation based on the language.
-
-In case the language is not recognized, the message will be dropped.
-In case the language is recognized but the it's neither english or french, it will continue the pipeline - in this case it will be written in the `default-topic`
+This example detects the language of the input record and then dispatches english and french records to different topics. 
+You could then implement another pipeline to perform different computations based on the language.
 
+If the language is not recognized, the message is dropped.
+If the language is recognized but it's neither english or french, the message will be written to the `default-topic`
 
 
 ### Configuration
 
-Checkout the full configuration properties in the [API Reference page](../../building-applications/api-reference/agents.md#dispatch).
+Check out the full configuration properties in the [API Reference page](../../building-applications/api-reference/agents.md#dispatch).

pipeline-agents/flow-control/timer-source.md

@@ -0,0 +1,36 @@
+# timer-source
+
+This source agent periodically emits a new record (an "event") and passes it to the pipeline.
+
+### Example
+
+This is an example using the `timer-source` agent to emit a new record every 60 seconds.
+
+```yaml
+  pipeline:
+    - name: "Timer"
+      type: "timer-source"    
+      configuration:
+        period-seconds: 60
+        fields:
+            - name: "key.id"
+              expression: "fn:uuid()"
+            - name: "value.string_payload"
+              expression: "'constant-string-payload'"
+            - name: "value.int_payload"
+              expression: "42"
+            - name: "properties.foo"
+              expression: "'some property'"
+```
+
+In this example the `timer-source` agent emits one record every 60 seconds.
+
+## Defining the contents of the output record
+
+The agent allows you to configure a set of fields that will be written to the output record.
+As usual you can write to the key part, the value part and the properties of the record.
+Use the [expression language](../../building-applications/expression-language.md) to define the fields and write the expression.
+
+### Configuration
+
+Checkout the full configuration properties in the [API Reference page](../../building-applications/api-reference/agents.md#timer-source).

pipeline-agents/flow-control/trigger-event.md

@@ -0,0 +1,49 @@
+# trigger-event
+
+This agent writes a new record (an "event") to a different topic, based on a condition.
+It can also optionally drop the record from the main flow of the pipeline.
+
+This agent is different from the [dispatch agent](dispatch.md) as the record that is sent to the new topic is a new record, not the same record that was received.
+
+### Example
+
+This is an example using the `trigger-event` agent to write a new record to a different topic when a condition is met.
+
+```yaml
+  - name: "Split some text"  
+    type: "text-splitter"
+    input: input-topic-splitter
+    configuration:
+       ....
+
+  - name: "Trigger event on last chunk"
+    type: "trigger-event"
+    output: output-topic-chunks
+    configuration:
+      destination: drop-stale-chunks-topic
+      continue-processing: true
+      when: fn:toInt(properties.text_num_chunks) == (fn:toInt(properties.chunk_id) + 1)
+      fields:
+          - name: "value.filename"
+            expression: "key.filename"
+```
+
+In this example the `text-splitter` agent splits a text into a set of chunks.
+When the last chunk is processed, the `trigger-event` agent writes a new record to the `drop-stale-chunks-topic` topic.
+The new record will have a filename field in the value part ("value.filename").
+
+## Defining the contents of the output record
+
+The `trigger-event` agent configures a set of fields that will be written to the output record.
+As usual you can write to the key part, the value part, and the properties of the record.
+Use the [expression language](../../building-applications/expression-language.md) to define the fields and write the expression.
+
+
+### Aborting the processing downstream
+
+The `trigger-event` agent can also abort the processing of the record downstream by setting the `continue-processing` property to `false`.
+This is useful in cases where you have some system events that you want to write to a different topic, but you don't want to continue processing the record downstream.
+
+### Configuration
+
+Check out the full configuration properties in the [API Reference page](../../building-applications/api-reference/agents.md#trigger-event).