import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters; //导入依赖的package包/类
private void createItemAndWait(
    UaSubscription.NotificationListener notificationListener,
    Object notificationLock) throws InterruptedException, ExecutionException {

    // create a subscription and a monitored item
    UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();
    subscription.addNotificationListener(notificationListener);

    ReadValueId readValueId = new ReadValueId(
        Identifiers.Server_ServerStatus_State,
        AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE);

    MonitoringParameters parameters = new MonitoringParameters(
        uint(1),    // client handle
        1000.0,     // sampling interval
        null,       // no (default) filter
        uint(10),   // queue size
        true);      // discard oldest

    MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
        readValueId, MonitoringMode.Reporting, parameters);

    synchronized (notificationLock) {
        subscription.createMonitoredItems(TimestampsToReturn.Both, newArrayList(request)).get();
        notificationLock.wait(5000);
    }
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters; //导入依赖的package包/类
@Test
public void testSubscribe() throws Exception {
    logger.info("testSubscribe()");

    // create a subscription and a monitored item
    UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();

    List<MonitoredItemCreateRequest> requests = new ArrayList<>();

    for (int i = 0; i < 10; i++) {
        ReadValueId readValueId = new ReadValueId(
            Identifiers.Server_ServerStatus_State,
            AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE);

        MonitoringParameters parameters = new MonitoringParameters(
            uint(i),    // client handle
            1000.0,     // sampling interval
            null,       // no (default) filter
            uint(10),   // queue size
            true);      // discard oldest

        MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
            readValueId, MonitoringMode.Reporting, parameters);

        requests.add(request);
    }

    @SuppressWarnings("unchecked")
    CompletableFuture<DataValue>[] cfs = new CompletableFuture[10];

    for (int i = 0; i < 10; i++) {
        CompletableFuture<DataValue> f = new CompletableFuture<>();
        f.thenAccept(value -> logger.info("received {}", value));
        cfs[i] = f;
    }

    BiConsumer<UaMonitoredItem, Integer> callback =
        (item, idx) -> item.setValueConsumer(cfs[idx]::complete);

    List<UaMonitoredItem> items = subscription
        .createMonitoredItems(TimestampsToReturn.Both, requests, callback).get();

    assertTrue(items.stream().allMatch(item -> item.getStatusCode().isGood()));
    assertNotNull(FutureUtils.sequence(cfs).get(5, TimeUnit.SECONDS));
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters; //导入依赖的package包/类
@Override
public void run(OpcUaClient client, CompletableFuture<OpcUaClient> future) throws Exception {
    // synchronous connect
    client.connect().get();

    // create a subscription and a monitored item
    UaSubscription subscription = client.getSubscriptionManager()
        .createSubscription(1000.0).get();

    ReadValueId readValueId = new ReadValueId(
        Identifiers.Server,
        AttributeId.EventNotifier.uid(),
        null,
        QualifiedName.NULL_VALUE
    );

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

    EventFilter eventFilter = new EventFilter(
        new SimpleAttributeOperand[]{
            new SimpleAttributeOperand(
                Identifiers.BaseEventType,
                new QualifiedName[]{new QualifiedName(0, "EventId")},
                AttributeId.Value.uid(),
                null),
            new SimpleAttributeOperand(
                Identifiers.BaseEventType,
                new QualifiedName[]{new QualifiedName(0, "Severity")},
                AttributeId.Value.uid(),
                null),
            new SimpleAttributeOperand(
                Identifiers.BaseEventType,
                new QualifiedName[]{new QualifiedName(0, "Time")},
                AttributeId.Value.uid(),
                null),
            new SimpleAttributeOperand(
                Identifiers.BaseEventType,
                new QualifiedName[]{new QualifiedName(0, "Message")},
                AttributeId.Value.uid(),
                null)
        },
        new ContentFilter(null)
    );

    MonitoringParameters parameters = new MonitoringParameters(
        clientHandle,
        0.0,
        ExtensionObject.encode(eventFilter),
        uint(10),
        true
    );

    MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
        readValueId,
        MonitoringMode.Reporting,
        parameters
    );

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

    // do something with the value updates
    UaMonitoredItem item = items.get(0);

    item.setEventConsumer((i, vs) -> {
        Arrays.stream(vs).forEach(
            v -> logger.info("{} variant received: {}",
                i.getReadValueId().getNodeId(), v.getValue())
        );

        future.complete(client);
    });
}