Scratch.java

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfigBuilder;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedDataItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedSubscription;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;

class Scratch {

    public static void main(String[] args) throws UaException, ExecutionException, InterruptedException {
        OpcUaClient client = OpcUaClient.create(
            "opc.tcp://milo.digitalpetri.com:62541/milo",
            endpoints ->
                endpoints.stream()
                    .filter(e -> e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri()))
                    .findFirst(),
            OpcUaClientConfigBuilder::build
        );

        client.connect().get();

        System.out.println("Connected");

        ManagedSubscription subscription = ManagedSubscription.create(client);

        List<NodeId> nodeIds = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            nodeIds.add(NodeId.parse("ns=2;i=" + i));
        }

        Map<ManagedDataItem, CompletableFuture<DataValue>> futures = new HashMap<>();

        List<ManagedDataItem> dataItems = subscription.createDataItems(
            nodeIds,
            dataItem -> {
                futures.put(dataItem, nextValueFuture(dataItem));

                dataItem.addDataValueListener(
                    value ->
                        System.out.printf("item=%s value=%s%n", dataItem.getNodeId(), value.getValue())
                );
            }
        );

        assert dataItems.stream().allMatch(item -> item.getStatusCode().isGood());

        System.out.println("Waiting for values...");

        CompletableFuture.allOf(futures.values().toArray(CompletableFuture[]::new)).get();

        System.out.println("Received all 10 values.");
    }

    private static CompletableFuture<DataValue> nextValueFuture(ManagedDataItem dataItem) {
        CompletableFuture<DataValue> future = new CompletableFuture<>();

        ManagedDataItem.DataValueListener listener =
            dataItem.addDataValueListener(future::complete);

        future.whenComplete((v, ex) -> dataItem.removeDataValueListener(listener));

        return future;
    }

}