import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的package包/类
@Override
public CompletableFuture<CreateMonitoredItemsResponse> createMonitoredItems(
    UInteger subscriptionId,
    TimestampsToReturn timestampsToReturn,
    List<MonitoredItemCreateRequest> itemsToCreate) {

    return getSession().thenCompose(session -> {
        CreateMonitoredItemsRequest request = new CreateMonitoredItemsRequest(
            newRequestHeader(session.getAuthenticationToken()),
            subscriptionId,
            timestampsToReturn,
            a(itemsToCreate, MonitoredItemCreateRequest.class));

        return sendRequest(request);
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的package包/类
@Override
public CompletableFuture<List<UaMonitoredItem>> createMonitoredItems(
    TimestampsToReturn timestampsToReturn,
    List<MonitoredItemCreateRequest> itemsToCreate,
    ItemCreationCallback itemCreationCallback) {

    return notificationSemaphore.acquire().thenCompose(permit -> {
        CompletableFuture<List<UaMonitoredItem>> itemsFuture =
            createMonitoredItems(timestampsToReturn, itemsToCreate);

        return itemsFuture.whenComplete((items, ex) -> {
            try {
                if (items != null) {
                    for (int i = 0; i < items.size(); i++) {
                        UaMonitoredItem item = items.get(i);

                        itemCreationCallback.onItemCreated(client.getDataTypeManager(), item, i);
                    }
                }
            } finally {
                permit.release();
            }
        });
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的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.MonitoredItemCreateRequest; //导入依赖的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.MonitoredItemCreateRequest; //导入依赖的package包/类
public PendingItemCreation(MonitoredItemCreateRequest request) {
    this.request = request;
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的package包/类
public MonitoredItemCreateRequest getRequest() {
    return request;
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的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);
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的package包/类
/**
 * Create one or more {@link UaMonitoredItem}s.
 * <p>
 * Callers must check the quality of each of the returned {@link UaMonitoredItem}s; it is not to be assumed that
 * all items were created successfully. Any item with a bad quality will not be updated nor will it be part of the
 * subscription's bookkeeping.
 * <p>
 * {@code itemCreationCallback} will be invoked for each successfully created {@link UaMonitoredItem}. Callers
 * should use this opportunity to register any value or event consumers on the item, as this is the only time in
 * which it is guaranteed no values or events will be delivered to the item yet.
 *
 * @param timestampsToReturn   the {@link TimestampsToReturn}.
 * @param itemsToCreate        a list of {@link MonitoredItemCreateRequest}s.
 * @param itemCreationCallback callback to be invoked for each successfully created {@link UaMonitoredItem}.
 * @return a list of {@link UaMonitoredItem}s.
 */
default CompletableFuture<List<UaMonitoredItem>> createMonitoredItems(
    TimestampsToReturn timestampsToReturn,
    List<MonitoredItemCreateRequest> itemsToCreate,
    BiConsumer<UaMonitoredItem, Integer> itemCreationCallback) {

    return createMonitoredItems(
        timestampsToReturn,
        itemsToCreate,
        (dataTypeManager, item, index) ->
            itemCreationCallback.accept(item, index)
    );
}
 

import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest; //导入依赖的package包/类
/**
 * This service is used to create and add one or more monitored items to a subscription.
 *
 * @param subscriptionId     the server-assigned identifier for the subscription that will report notifications for
 *                           the monitored items.
 * @param timestampsToReturn the {@link TimestampsToReturn}.
 * @param itemsToCreate      a list of monitored items to be created and assigned to the specified subscription.
 * @return a {@link CompletableFuture} containing the {@link CreateMonitoredItemsResponse}.
 */
CompletableFuture<CreateMonitoredItemsResponse> createMonitoredItems(
    UInteger subscriptionId,
    TimestampsToReturn timestampsToReturn,
    List<MonitoredItemCreateRequest> itemsToCreate);