ballerina 学习十 streams

Posted 2020-11-06 rongfengliang-荣锋亮

ballerina 的streams 使用的是siddhi complex event processing 引擎处理，可以包含的语法有
projection filtering windows join pattern

简单例子

• 代码

import ballerina/io;
import ballerina/runtime;
type StatusCount {
string status;
int totalCount;
};

type Teacher {
string name;
int age;
string status;
string batch;
string school;
};
function testAggregationQuery(
stream<StatusCount> filteredStatusCountStream,
stream<Teacher> teacherStream) {
forever {

//  cep  处理并发布结果
from teacherStream where age > 18 window lengthBatch(3)
select status, count(status) as totalCount
group by status
having totalCount > 1
=> (StatusCount[] status) {
filteredStatusCountStream.publish(status);
}
}
}

function main(string… args) {
stream<StatusCount> filteredStatusCountStream;

stream<Teacher> teacherStream;

testAggregationQuery(filteredStatusCountStream, teacherStream);

Teacher t1 = {name: "Sam", age: 25, status: "single",
batch: "LK2014", school: "Hampden High School"};
Teacher t2 = {name: "Jordan", age: 33, status: "single",
batch: "LK1998", school: "Columbia High School"};
Teacher t3 = {name: "Morgan", age: 45, status: "married",
batch: "LK1988", school: "Central High School"};

filteredStatusCountStream.subscribe(printStatusCount);

// 生产数据
teacherStream.publish(t1);
teacherStream.publish(t2);
teacherStream.publish(t3);

runtime:sleep(1000);
}
function printStatusCount(StatusCount s) {
io:println("Event received; status: " + s.status +
", total occurrences: " + s.totalCount);
}

• 输出结果

Event received; status: single, total occurrences: 2

• 用途
  对于事件处理的应用特别方便，比如日志处理，以及响应式系统开发，其中的siddhi 也是一个很不错的工具
• 参考资料
  https://ballerina.io/learn/by-example/hello-world-streams.html