环境:WSO2 流处理器 4.3.0
假设我有两个非常简单的流:
实时交付新创建的请求(未完成)的流(t1)
RequestStream(requestId)
实时完成请求时出现 requestsIds 的流 (t2)
FulfilmentStream(requestId)
保证 t2 总是 > t1
如何实现一个 SiddhiQL 语句来识别出现在RequestStream (event1)中并且在event1 过去 5 分钟后没有出现在FulfilmentStream (event2)中的 requestIds?
基于 Tishan 的工作 Siddhi App 答案:
@App:name('FailedToFulfillInAmountOfTime')
@source(
type="kafka",
topic.list="some_topic",
threading.option="single.thread",
group.id="some_group",
bootstrap.servers="xxx.xxx.xxx.xxx:6667",
@Map(type="json", @attributes(request_id = '$.alarm_id', severity = '$.severity', managed_object = '$.ManagedObject')))
define stream OrigAlarmStream (request_id int, severity string, managed_object string);
@sink(type='log', prefix='Got this execution request')
define stream RequestStream (request_id int, severity string, managed_object string);
@sink(type='log', prefix='Got this fulfillment confirmation:')
define stream FulfillmentStream (request_id int, severity string, managed_object string);
@sink(type='log', prefix='This fulfillment was not done within 1 min:')
define stream AlertStream(request_id int);
@info(name='getExpiredRequests')
from every e1=RequestStream -> not FulfillmentStream[e1.request_id == request_id] for 1 min
select e1.request_id
insert into AlertStream;
@info(name='CopyFulfillments')
from OrigAlarmStream[severity == 'Clear']
select request_id, severity, managed_object
insert into FulfillmentStream;
@info(name='CopyRequests')
from OrigAlarmStream[severity != 'Clear']
select request_id, severity, managed_object
insert into RequestStream;