Writing Ballerina Streaming Extensions
Custom Ballerina streaming extensions can be written in order to apply use case specific logic that is not available in Ballerina Streaming out of the box. There are two types of Ballerina streaming extensions that you can write to cater to your specific streaming use cases. They are;
- Window Extension
- Aggregator Extension
Writing Window Extensions
Window Extension allows events to be collected and expired without altering the event format based on the given input parameters. To write a custom Window extension, introduce an object type implementing “Window” abstract object of “streams” package. This Window type consists of two methods like below;
<Window>:process(…) — Method to implement the actual window logic when an event arrives in the Window.
<Window>:getCandidateEvents(…) — Method to return candidate events in a join scenario.
For example, Window extension created with the function name “length” can be referred in a query as follows:
A sample implementation of a LengthWindow extension can be found below;
Writing Aggregator Extensions
Aggregator extensions consume zero or more parameters for each event and output a single attribute having an aggregated results based on the input parameters as an output. These aggregator extensions can be used with conjunction with a window in order to find the aggregated results based on a given window.
To write a custom aggregator, introduce an object type implementing “Aggregator” abstract object of “streams” package. This Aggregator type consists of two methods like below;
<Aggregator>:process(…) — Method to implement the actual aggregate logic when an event arrives in the Aggregator.
<Aggregator>:copy(…)— Method to create and return a new aggregator instance.
For example, the Aggregator extension created with the function name “sum” can be referred in a query as follows:
A sample implementation of a “sum” Aggregator extension can be found below;
If you find this post useful, Please share or leave a comment… 🙂
