import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
private void pollForData() {
  if (destroyed) {
    return;
  }
  refreshNodeIds();
  opcUaClient.readValues(0.0, TimestampsToReturn.Both, nodeIds)
      .thenAccept(values -> {
        try {
          process(values, System.currentTimeMillis() + "." + counter.getAndIncrement(), null);
        } catch (Exception ex) {
          LOG.error(Errors.OPC_UA_09.getMessage(), ex.getMessage(), ex);
          errorQueue.offer(new StageException(Errors.OPC_UA_09, ex.getMessage(), ex));
        }
        if (ThreadUtil.sleep(conf.pollingInterval)) {
          pollForData();
        }
      })
      .exceptionally(ex -> {
        LOG.warn(Errors.OPC_UA_12.getMessage(), ex.getMessage());
        reConnect();
        pollForData();
        return null;
      });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public void read(
        final ReadContext context,
        final Double maxAge,
        final TimestampsToReturn timestamps,
        final List<ReadValueId> readValueIds) {

    final List<DataValue> results = new ArrayList<>(readValueIds.size());

    for (final ReadValueId id : readValueIds) {
        final ServerNode node = this.nodeMap.get(id.getNodeId());

        final DataValue value = node != null
                ? node.readAttribute(new AttributeContext(context), id.getAttributeId())
                : new DataValue(StatusCodes.Bad_NodeIdUnknown);

        results.add(value);
    }

    // report back with result

    context.complete(results);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
public CompletableFuture<ReadResponse> testStack(NodeId nodeId) {
    RequestHeader header = new RequestHeader(
        NodeId.NULL_VALUE,
        DateTime.now(),
        uint(requestHandle.getAndIncrement()),
        uint(0),
        null,
        uint(60000),
        null
    );

    ReadRequest request = new ReadRequest(
        header,
        0.0,
        TimestampsToReturn.Neither,
        new ReadValueId[]{
            new ReadValueId(
                nodeId,
                AttributeId.Value.uid(),
                null,
                null)
        }
    );

    return client.sendRequest(request);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public CompletableFuture<HistoryReadResponse> historyRead(HistoryReadDetails historyReadDetails,
                                                          TimestampsToReturn timestampsToReturn,
                                                          boolean releaseContinuationPoints,
                                                          List<HistoryReadValueId> nodesToRead) {

    return getSession().thenCompose(session -> {
        HistoryReadRequest request = new HistoryReadRequest(
            newRequestHeader(session.getAuthenticationToken()),
            ExtensionObject.encode(historyReadDetails),
            timestampsToReturn,
            releaseContinuationPoints,
            a(nodesToRead, HistoryReadValueId.class));

        return sendRequest(request);
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的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.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public CompletableFuture<ModifyMonitoredItemsResponse> modifyMonitoredItems(
    UInteger subscriptionId,
    TimestampsToReturn timestampsToReturn,
    List<MonitoredItemModifyRequest> itemsToModify) {

    return getSession().thenCompose(session -> {
        ModifyMonitoredItemsRequest request = new ModifyMonitoredItemsRequest(
            newRequestHeader(session.getAuthenticationToken()),
            subscriptionId,
            timestampsToReturn,
            a(itemsToModify, MonitoredItemModifyRequest.class));

        return sendRequest(request);
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
/**
 * This service is used to read one or more attributes of one or more nodes.
 *
 * @param maxAge             the requested max age of the value, in milliseconds. If maxAge is set to 0, the Server
 *                           shall attempt to read a new value from the data source. If maxAge is set to the max
 *                           Int32 value or greater, the Server shall attempt to get a cached value. Negative values
 *                           are invalid for maxAge.
 * @param timestampsToReturn the requested {@link TimestampsToReturn}.
 * @param nodeIds            the {@link NodeId}s identifying the nodes to read.
 * @param attributeIds       the attribute ids to read, the size and order matching the provided {@link NodeId}s.
 * @return a {@link CompletableFuture} containing a list of {@link DataValue}s, the size and order matching the
 * provided {@link NodeId}s.
 */
default CompletableFuture<List<DataValue>> read(double maxAge,
                                                TimestampsToReturn timestampsToReturn,
                                                List<NodeId> nodeIds,
                                                List<UInteger> attributeIds) {

    if (nodeIds.size() != attributeIds.size()) {
        CompletableFuture<List<DataValue>> failed = new CompletableFuture<>();
        failed.completeExceptionally(new IllegalArgumentException("nodeIds.size() != attributeIds.size()"));
        return failed;
    } else {
        Stream<ReadValueId> stream = StreamUtils.zip(
            nodeIds.stream(), attributeIds.stream(),
            (nId, aId) -> new ReadValueId(nId, aId, null, QualifiedName.NULL_VALUE));

        return read(maxAge, timestampsToReturn, stream.collect(Collectors.toList()))
            .thenApply(r -> l(r.getResults()));
    }
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的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.enumerated.TimestampsToReturn; //导入依赖的package包/类
protected CompletableFuture<DataValue> readAttribute(AttributeId attributeId) {
    Optional<DataValue> opt =
        AttributeId.BASE_NODE_ATTRIBUTES.contains(attributeId) ?
            nodeCache.getAttribute(nodeId, attributeId) : Optional.empty();

    return opt.map(CompletableFuture::completedFuture).orElseGet(() -> {
        ReadValueId readValueId = new ReadValueId(
            nodeId, attributeId.uid(), null, QualifiedName.NULL_VALUE);

        CompletableFuture<ReadResponse> future =
            client.read(0.0, TimestampsToReturn.Both, newArrayList(readValueId));

        return future.thenApply(response -> {
            DataValue value = l(response.getResults()).get(0);

            if (attributeId != AttributeId.Value) {
                nodeCache.putAttribute(nodeId, attributeId, value);
            }

            return value;
        });
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public CompletableFuture<UaNode> createNode(NodeId nodeId) {
    ReadValueId readValueId = new ReadValueId(
        nodeId, AttributeId.NodeClass.uid(), null, QualifiedName.NULL_VALUE);

    CompletableFuture<ReadResponse> future =
        client.read(0.0, TimestampsToReturn.Neither, newArrayList(readValueId));

    return future.thenCompose(response -> {
        DataValue value = l(response.getResults()).get(0);
        NodeClass nodeClass = NodeClass.from((Integer) value.getValue().getValue());

        if (nodeClass != null) {
            client.getNodeCache().putAttribute(nodeId, AttributeId.NodeClass, value);

            return completedFuture(createNode(nodeId, nodeClass));
        } else {
            return failedFuture(new UaException(value.getStatusCode(), "NodeClass was null"));
        }
    });
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
private CompletableFuture<List<DataValue>> readDataAttributes(Session session, Namespace namespace, NodeId itemId) {
    Function<AttributeId, ReadValueId> f = id ->
        new ReadValueId(itemId, id.uid(), null, QualifiedName.NULL_VALUE);

    CompletableFuture<List<DataValue>> future = new CompletableFuture<>();

    ReadContext readContext = new ReadContext(
        server, session, future, new DiagnosticsContext<>());

    List<ReadValueId> attributes = newArrayList(
        f.apply(AttributeId.AccessLevel),
        f.apply(AttributeId.UserAccessLevel),
        f.apply(AttributeId.MinimumSamplingInterval));

    namespace.read(readContext, 0.0, TimestampsToReturn.Neither, attributes);

    return future;
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public void read(
    ReadContext context,
    Double maxAge,
    TimestampsToReturn timestamps,
    List<ReadValueId> readValueIds) {

    List<DataValue> results = Lists.newArrayListWithCapacity(readValueIds.size());

    for (ReadValueId id : readValueIds) {
        ServerNode node = nodeMap.get(id.getNodeId());

        DataValue value = (node != null) ?
            node.readAttribute(new AttributeContext(context), id.getAttributeId()) :
            new DataValue(StatusCodes.Bad_NodeIdUnknown);

        results.add(value);
    }

    context.complete(results);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的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.TimestampsToReturn; //导入依赖的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.TimestampsToReturn; //导入依赖的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.TimestampsToReturn; //导入依赖的package包/类
public synchronized void modify(TimestampsToReturn timestamps,
                                UInteger clientHandle,
                                double samplingInterval,
                                ExtensionObject filter,
                                UInteger queueSize,
                                boolean discardOldest) throws UaException {

    installFilter(filter);

    this.timestamps = timestamps;
    this.clientHandle = clientHandle.longValue();
    this.samplingInterval = samplingInterval;
    this.discardOldest = discardOldest;

    if (queueSize.intValue() != this.queueSize) {
        setQueueSize(queueSize);

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

        while (oldQueue.size() > 0) {
            enqueue(oldQueue.remove());
        }
    }
}
 

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

    DataValue dataValue = client.readValue(
        0.0,
        TimestampsToReturn.Neither,
        NodeId.parse("ns=2;s=Demo.Static.Scalar.WorkOrder")
    ).get();

    ExtensionObject xo = (ExtensionObject) dataValue.getValue().getValue();

    // Decoding a struct with custom DataType requires a DataTypeManager
    // that has the codec registered with it. OpcUaClient automatically
    // reads any DataTypeDictionary nodes present in the server upon
    // connecting and dynamically generates codecs for custom structures.
    Object value = xo.decode(client.getDataTypeManager());

    logger.info("value: {}", value);

    future.complete(client);
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
/**
 * Derive a new {@link DataValue} from a given {@link DataValue}.
 *
 * @param from       the {@link DataValue} to derive from.
 * @param timestamps the timestamps to return in the derived value.
 * @return a derived {@link DataValue}.
 */
public static DataValue derivedValue(DataValue from, TimestampsToReturn timestamps) {
    boolean includeSource = timestamps == TimestampsToReturn.Source || timestamps == TimestampsToReturn.Both;
    boolean includeServer = timestamps == TimestampsToReturn.Server || timestamps == TimestampsToReturn.Both;

    return new DataValue(
        from.value,
        from.status,
        includeSource ? from.sourceTime : null,
        includeServer ? from.serverTime : null
    );
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
/**
 * Derive a new {@link DataValue} from a given {@link DataValue}.
 * <p>
 * The value is assumed to be for a non-value Node attribute, and therefore the source timestamp is not returned.
 *
 * @param from       the {@link DataValue} to derive from.
 * @param timestamps the timestamps to return in the derived value.
 * @return a derived {@link DataValue}.
 */
public static DataValue derivedNonValue(DataValue from, TimestampsToReturn timestamps) {
    boolean includeServer = timestamps == TimestampsToReturn.Server || timestamps == TimestampsToReturn.Both;

    return new DataValue(
        from.value,
        from.status,
        null,
        includeServer ? from.serverTime : null
    );
}
 

import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; //导入依赖的package包/类
@Override
public CreateMonitoredItemsRequest decode(UaDecoder decoder) throws UaSerializationException {
    RequestHeader requestHeader = (RequestHeader) decoder.readBuiltinStruct("RequestHeader", RequestHeader.class);
    UInteger subscriptionId = decoder.readUInt32("SubscriptionId");
    TimestampsToReturn timestampsToReturn = TimestampsToReturn.from(decoder.readInt32("TimestampsToReturn"));
    MonitoredItemCreateRequest[] itemsToCreate =
        decoder.readBuiltinStructArray(
            "ItemsToCreate",
            MonitoredItemCreateRequest.class
        );

    return new CreateMonitoredItemsRequest(requestHeader, subscriptionId, timestampsToReturn, itemsToCreate);
}