Milo-OPC UA处理Subscription和Triggering

Subscription有两种模式，一种是Reporting，另一种是Sampling。

如果定义为Sampling，则这个Subscription是一个Triggered Item，即被激发的订阅，需要一个定义为Reporting的Subscription（称为Triggering Item）与它连接。这样当Triggering Item更新时，会激发Triggered Item更新。代码如下：

public class TriggeringExample implements UaClient {

// public static void main(String[] args) throws Exception {
// TriggeringExample example = new TriggeringExample();
//
// new ClientExampleRunner(example).run();
// }

private final Logger logger = LoggerFactory.getLogger(getClass());

private final AtomicLong clientHandles = new AtomicLong(1L);

@Override
public void run(OpcUaClient client, CompletableFuture<OpcUaClient> future) throws Exception {
// synchronous connect
client.connect().get();

// create a subscription @ 1000ms  一个订阅可以包含多个监控item
UaSubscription subscription = client.getSubscriptionManager()
.createSubscription(1000.0)
.get();

// subscribe to a static value that reports
ReadValueId readValueId1 = new ReadValueId(
new NodeId(2, "ch1.d1.tag1"),
AttributeId.Value.uid(),
null,
QualifiedName.NULL_VALUE
);

//创建监控item, 第一个为Reporting mode

UaMonitoredItem reportingItem = createMonitoredItem(subscription, readValueId1, MonitoringMode.Reporting);

//创建第二个监控item， 它是Sampling mode，需要第一个项激发
ReadValueId readValueId2 = new ReadValueId(
new NodeId(2, "ch1.d1.tag2"),
AttributeId.Value.uid(),
null,
QualifiedName.NULL_VALUE
);


UaMonitoredItem samplingItem = createMonitoredItem(subscription, readValueId2, MonitoringMode.Sampling);

 

//将Triggering Item与Triggered item（可以有多个）连接起来（注意这里用了异步模式）

subscription.addTriggeringLinks(reportingItem, newArrayList(samplingItem)).get();

// trigger reporting of both by writing to the static item and changing its value
//VariableNode node = client.getAddressSpace().createVariableNode(
// new NodeId(2, "ch1.d1.tag1"));

//通过改变Triggering Item来激发Triggered item向我发送消息，注意这里writeValue使用了异步模式，如果没有这个get()，只能得到一两次消息更新
for(int i=0; i<10000; i++){
Variant newVal = new Variant(Unsigned.ushort(i));
DataValue va = new DataValue(newVal, null, null);
client.writeValue(new NodeId(2, "ch1.d1.tag1"), va).get();

}
//node.writeValue(va);

// client.writeValue(
// new NodeId(2, "ch1.d1.tag1"),
// va
// ).get();

// let the example run for 5 seconds then terminate
Thread.sleep(500000);
future.complete(client);
}

private UaMonitoredItem createMonitoredItem(
UaSubscription subscription,
ReadValueId readValueId,
MonitoringMode monitoringMode
) throws ExecutionException, InterruptedException {

// important: client handle must be unique per item
UInteger clientHandle = uint(clientHandles.getAndIncrement());

MonitoringParameters parameters = new MonitoringParameters(
clientHandle,
1000.0,
null,
uint(10),
true
);

MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
readValueId,
monitoringMode,
parameters
);

BiConsumer<UaMonitoredItem, Integer> onItemCreated =
(item, id) -> item.setValueConsumer(this::onSubscriptionValue);

List<UaMonitoredItem> items = subscription.createMonitoredItems(
TimestampsToReturn.Both,
newArrayList(request),
onItemCreated
).get();

return items.get(0);
}

private void onSubscriptionValue(UaMonitoredItem item, DataValue value) {
logger.info(
"回调：subscription value received: item={}, value={}",
item.getReadValueId().getNodeId(), value.getValue());
}

}