import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public OpcUaMonitoredItem(
    OpcUaClient client,
    UInteger clientHandle,
    ReadValueId readValueId,
    UInteger monitoredItemId,
    StatusCode statusCode,
    double revisedSamplingInterval,
    UInteger revisedQueueSize,
    ExtensionObject filterResult,
    MonitoringMode monitoringMode,
    ExtensionObject monitoringFilter) {

    this.client = client;
    this.clientHandle = clientHandle;
    this.readValueId = readValueId;
    this.monitoredItemId = monitoredItemId;
    this.statusCode = statusCode;
    this.revisedSamplingInterval = revisedSamplingInterval;
    this.revisedQueueSize = revisedQueueSize;
    this.filterResult = filterResult;
    this.monitoringMode = monitoringMode;
    this.monitoringFilter = monitoringFilter;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public MonitoredDataItem(
    UInteger id,
    UInteger subscriptionId,
    ReadValueId readValueId,
    MonitoringMode monitoringMode,
    TimestampsToReturn timestamps,
    UInteger clientHandle,
    double samplingInterval,
    ExtensionObject filter,
    UInteger queueSize,
    boolean discardOldest) throws UaException {

    super(id, subscriptionId, readValueId, monitoringMode,
        timestamps, clientHandle, samplingInterval, queueSize, discardOldest);

    installFilter(filter);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public MonitoredEventItem(
    UInteger id,
    UInteger subscriptionId,
    ReadValueId readValueId,
    MonitoringMode monitoringMode,
    TimestampsToReturn timestamps,
    UInteger clientHandle,
    double samplingInterval,
    UInteger queueSize,
    boolean discardOldest,
    ExtensionObject filter) throws UaException {

    super(id, subscriptionId, readValueId, monitoringMode,
        timestamps, clientHandle, samplingInterval, queueSize, discardOldest);

    installFilter(filter);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
protected BaseMonitoredItem(
    UInteger id,
    UInteger subscriptionId,
    ReadValueId readValueId,
    MonitoringMode monitoringMode,
    TimestampsToReturn timestamps,
    UInteger clientHandle,
    double samplingInterval,
    UInteger queueSize,
    boolean discardOldest) {

    this.id = id;
    this.subscriptionId = subscriptionId;
    this.readValueId = readValueId;
    this.monitoringMode = monitoringMode;
    this.timestamps = timestamps;
    this.clientHandle = clientHandle.longValue();
    this.samplingInterval = samplingInterval;
    this.discardOldest = discardOldest;

    setQueueSize(queueSize);

    queue = new RingBuffer<>(this.queueSize);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public SetMonitoringModeRequest decode(UaDecoder decoder) throws UaSerializationException {
    RequestHeader requestHeader = (RequestHeader) decoder.readBuiltinStruct("RequestHeader", RequestHeader.class);
    UInteger subscriptionId = decoder.readUInt32("SubscriptionId");
    MonitoringMode monitoringMode = MonitoringMode.from(decoder.readInt32("MonitoringMode"));
    UInteger[] monitoredItemIds = decoder.readArray("MonitoredItemIds", decoder::readUInt32, UInteger.class);

    return new SetMonitoringModeRequest(requestHeader, subscriptionId, monitoringMode, monitoredItemIds);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public MonitoredItemCreateRequest decode(UaDecoder decoder) throws UaSerializationException {
    ReadValueId itemToMonitor = (ReadValueId) decoder.readBuiltinStruct("ItemToMonitor", ReadValueId.class);
    MonitoringMode monitoringMode = MonitoringMode.from(decoder.readInt32("MonitoringMode"));
    MonitoringParameters requestedParameters = (MonitoringParameters) decoder.readBuiltinStruct("RequestedParameters", MonitoringParameters.class);

    return new MonitoredItemCreateRequest(itemToMonitor, monitoringMode, requestedParameters);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的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.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public CompletableFuture<SetMonitoringModeResponse> setMonitoringMode(UInteger subscriptionId,
                                                                      MonitoringMode monitoringMode,
                                                                      List<UInteger> monitoredItemIds) {

    return getSession().thenCompose(session -> {
        SetMonitoringModeRequest request = new SetMonitoringModeRequest(
            newRequestHeader(session.getAuthenticationToken()),
            subscriptionId,
            monitoringMode,
            a(monitoredItemIds, UInteger.class));

        return sendRequest(request);
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public CompletableFuture<List<StatusCode>> setMonitoringMode(MonitoringMode monitoringMode,
                                                             List<UaMonitoredItem> items) {

    List<UInteger> monitoredItemIds = items.stream()
        .map(UaMonitoredItem::getMonitoredItemId)
        .collect(Collectors.toList());

    CompletableFuture<SetMonitoringModeResponse> future =
        client.setMonitoringMode(subscriptionId, monitoringMode, monitoredItemIds);

    return future.thenApply(response -> {
        List<StatusCode> results = l(response.getResults());

        for (int i = 0; i < monitoredItemIds.size(); i++) {
            UInteger id = monitoredItemIds.get(i);
            OpcUaMonitoredItem item = itemsByServerHandle.get(id);

            StatusCode result = results.get(i);
            if (result.isGood() && item != null) {
                item.setMonitoringMode(monitoringMode);
            }
        }

        return results;
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public synchronized void setMonitoringMode(MonitoringMode monitoringMode) {
    if (monitoringMode == MonitoringMode.Disabled) {
        lastValue = null;
    }

    super.setMonitoringMode(monitoringMode);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public void setMonitoringMode(MonitoringMode monitoringMode) {
    this.monitoringMode = monitoringMode;

    if (monitoringMode == MonitoringMode.Disabled) {
        queue.clear();
    }
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public SetMonitoringModeRequest(RequestHeader requestHeader, UInteger subscriptionId, MonitoringMode monitoringMode, UInteger[] monitoredItemIds) {
    this.requestHeader = requestHeader;
    this.subscriptionId = subscriptionId;
    this.monitoringMode = monitoringMode;
    this.monitoredItemIds = monitoredItemIds;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public MonitoredItemCreateRequest(ReadValueId itemToMonitor, MonitoringMode monitoringMode, MonitoringParameters requestedParameters) {
    this.itemToMonitor = itemToMonitor;
    this.monitoringMode = monitoringMode;
    this.requestedParameters = requestedParameters;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的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.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public MonitoringMode getMonitoringMode() {
    return monitoringMode;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
void setMonitoringMode(MonitoringMode monitoringMode) {
    this.monitoringMode = monitoringMode;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public void setMonitoringMode(ServiceRequest<SetMonitoringModeRequest, SetMonitoringModeResponse> service) {
    SetMonitoringModeRequest request = service.getRequest();
    UInteger subscriptionId = request.getSubscriptionId();

    try {
        Subscription subscription = subscriptions.get(subscriptionId);
        List<UInteger> itemsToModify = l(request.getMonitoredItemIds());

        if (subscription == null) {
            throw new UaException(StatusCodes.Bad_SubscriptionIdInvalid);
        }
        if (itemsToModify.isEmpty()) {
            throw new UaException(StatusCodes.Bad_NothingToDo);
        }

        /*
         * Set MonitoringMode on each monitored item, if it exists.
         */

        MonitoringMode monitoringMode = request.getMonitoringMode();
        StatusCode[] results = new StatusCode[itemsToModify.size()];
        List<BaseMonitoredItem<?>> modified = newArrayListWithCapacity(itemsToModify.size());

        for (int i = 0; i < itemsToModify.size(); i++) {
            UInteger itemId = itemsToModify.get(i);
            BaseMonitoredItem<?> item = subscription.getMonitoredItems().get(itemId);

            if (item != null) {
                item.setMonitoringMode(monitoringMode);

                modified.add(item);

                results[i] = StatusCode.GOOD;
            } else {
                results[i] = new StatusCode(StatusCodes.Bad_MonitoredItemIdInvalid);
            }
        }

        /*
         * Notify namespaces of the items whose MonitoringMode has been modified.
         */

        Map<UShort, List<MonitoredItem>> byNamespace = modified.stream()
            .collect(Collectors.groupingBy(item -> item.getReadValueId().getNodeId().getNamespaceIndex()));

        byNamespace.keySet().forEach(namespaceIndex -> {
            List<MonitoredItem> items = byNamespace.get(namespaceIndex);
            server.getNamespaceManager().getNamespace(namespaceIndex).onMonitoringModeChanged(items);
        });

        /*
         * Build and return results.
         */

        ResponseHeader header = service.createResponseHeader();
        SetMonitoringModeResponse response = new SetMonitoringModeResponse(
            header, results, new DiagnosticInfo[0]);

        service.setResponse(response);
    } catch (UaException e) {
        service.setServiceFault(e);
    }
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
@Override
public boolean isSamplingEnabled() {
    return getMonitoringMode() != MonitoringMode.Disabled;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public synchronized boolean hasNotifications() {
    return (queue.size() > 0 && monitoringMode == MonitoringMode.Reporting);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode; //导入依赖的package包/类
public MonitoringMode getMonitoringMode() {
    return monitoringMode;
}