SubscriptionExample.java

/*
 * Copyright (c) 2019 the Eclipse Milo Authors
 *
 * This program and the accompanying materials are made
 * available under the terms of the Eclipse Public License 2.0
 * which is available at https://www.eclipse.org/legal/epl-2.0/
 *
 * SPDX-License-Identifier: EPL-2.0
 */

package org.eclipse.milo.examples.client;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.BiConsumer;

import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.SessionActivityListener;
import org.eclipse.milo.opcua.sdk.client.api.UaSession;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscriptionManager;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.google.common.collect.Lists.newArrayList;
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;

public class SubscriptionExample implements ClientExample {

    public static void main(String[] args) throws Exception {
        SubscriptionExample example = new SubscriptionExample();

        new ClientExampleRunner(example, false).run();
    }

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public String getEndpointUrl() {
        return "opc.tcp://milo.digitalpetri.com:62541/milo";
    }

    @Override
    public void run(OpcUaClient client, CompletableFuture<OpcUaClient> future) throws Exception {
        // synchronous connect
        client.connect().get();

        client.addSessionActivityListener(new SessionActivityListener() {
            @Override
            public void onSessionActive(UaSession session) {
                logger.info("onSessionActive: {}", session.getSessionId());
            }

            @Override
            public void onSessionInactive(UaSession session) {
                logger.info("onSessionInactive: {}", session.getSessionId());
            }
        });

        client.getSubscriptionManager().addSubscriptionListener(new UaSubscriptionManager.SubscriptionListener() {
            @Override
            public void onSubscriptionTransferFailed(UaSubscription subscription, StatusCode statusCode) {
                logger.warn("transfer for subscriptionId={} failed: {}", subscription.getSubscriptionId(), statusCode);
                try {
                    createSubscription(client);
                } catch (InterruptedException | ExecutionException e) {
                    logger.error("re-creating subscription failed", e);
                }
            }
        });

        createSubscription(client);

        // let the example run for 5 seconds then terminate
        Thread.sleep(Integer.MAX_VALUE);
        future.complete(client);
    }

    private void createSubscription(OpcUaClient client) throws InterruptedException, java.util.concurrent.ExecutionException {
        // create a subscription @ 1000ms
        UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();

        // subscribe to the Value attribute of the server's CurrentTime node
        ReadValueId readValueId = new ReadValueId(
            Identifiers.Server_ServerStatus_CurrentTime,
            AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE
        );

        // IMPORTANT: client handle must be unique per item within the context of a subscription.
        // You are not required to use the UaSubscription's client handle sequence; it is provided as a convenience.
        // Your application is free to assign client handles by whatever means necessary.
        UInteger clientHandle = subscription.nextClientHandle();

        MonitoringParameters parameters = new MonitoringParameters(
            clientHandle,
            1000.0,     // sampling interval
            null,       // filter, null means use default
            uint(10),   // queue size
            true        // discard oldest
        );

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

        // when creating items in MonitoringMode.Reporting this callback is where each item needs to have its
        // value/event consumer hooked up. The alternative is to create the item in sampling mode, hook up the
        // consumer after the creation call completes, and then change the mode for all items to reporting.
        BiConsumer<UaMonitoredItem, Integer> onItemCreated =
            (item, id) -> item.setValueConsumer(this::onSubscriptionValue);

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

        for (UaMonitoredItem item : items) {
            if (item.getStatusCode().isGood()) {
                logger.info("item created for nodeId={}", item.getReadValueId().getNodeId());
            } else {
                logger.warn(
                    "failed to create item for nodeId={} (status={})",
                    item.getReadValueId().getNodeId(), item.getStatusCode());
            }
        }
    }

    private void onSubscriptionValue(UaMonitoredItem item, DataValue value) {
        logger.info(
            "subscription value received: item={}, value={}",
            item.getReadValueId().getNodeId(), value.getValue());
    }

}

Logs:

12:30:57.066 [main] INFO  o.e.m.e.client.SubscriptionExample - item created for nodeId=NodeId{ns=0, id=2258}
12:30:58.129 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610570450000, javaDate=Wed May 13 12:30:57 EDT 2020}}
12:30:59.023 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610580450000, javaDate=Wed May 13 12:30:58 EDT 2020}}
12:31:00.171 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610590450000, javaDate=Wed May 13 12:30:59 EDT 2020}}
12:31:01.020 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610600460000, javaDate=Wed May 13 12:31:00 EDT 2020}}
12:31:02.024 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610610460000, javaDate=Wed May 13 12:31:01 EDT 2020}}
12:31:03.025 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610620470000, javaDate=Wed May 13 12:31:02 EDT 2020}}
12:31:04.063 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610630470000, javaDate=Wed May 13 12:31:03 EDT 2020}}
12:31:05.087 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610640480000, javaDate=Wed May 13 12:31:04 EDT 2020}}
12:31:06.110 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610650480000, javaDate=Wed May 13 12:31:05 EDT 2020}}
12:31:07.024 [milo-shared-thread-pool-1] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610660490000, javaDate=Wed May 13 12:31:06 EDT 2020}}
12:31:08.056 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610670490000, javaDate=Wed May 13 12:31:07 EDT 2020}}
12:31:09.080 [milo-shared-thread-pool-1] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610680490000, javaDate=Wed May 13 12:31:08 EDT 2020}}
12:31:10.212 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610690500000, javaDate=Wed May 13 12:31:09 EDT 2020}}
12:31:11.026 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338610700500000, javaDate=Wed May 13 12:31:10 EDT 2020}}
12:31:22.010 [milo-shared-thread-pool-1] WARN  o.e.milo.opcua.sdk.client.SessionFsm - [2] Keep Alive failureCount=2 exceeds failuresAllowed=1
12:31:22.013 [milo-shared-thread-pool-2] INFO  o.e.m.e.client.SubscriptionExample - onSessionInactive: NodeId{ns=1, id=Session:705dc048-da46-4ec0-a8d5-08cf67ef3b32}
12:31:27.010 [milo-shared-thread-pool-1] WARN  o.e.milo.opcua.sdk.client.SessionFsm - [2] Keep Alive failureCount=3 exceeds failuresAllowed=1
12:31:51.728 [milo-shared-thread-pool-2] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=23 for PublishResponse
12:31:51.906 [milo-shared-thread-pool-2] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=24 for PublishResponse
12:31:51.907 [milo-shared-thread-pool-2] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=26 for ReadResponse
12:31:51.908 [milo-shared-thread-pool-1] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=25 for ReadResponse
12:31:51.909 [milo-shared-thread-pool-1] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=27 for ReadResponse
12:31:51.940 [milo-shared-thread-pool-1] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=28 for CreateSessionResponse
12:31:51.964 [milo-shared-thread-pool-1] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=29 for CreateSessionResponse
12:31:51.992 [milo-shared-thread-pool-1] WARN  o.e.m.o.s.c.t.u.UascClientMessageHandler - No pending request with requestId=30 for CreateSessionResponse
12:31:52.522 [milo-shared-thread-pool-1] WARN  o.e.m.e.client.SubscriptionExample - transfer for subscriptionId=5 failed: StatusCode{name=Bad_UserAccessDenied, value=0x801F0000, quality=bad}
12:31:52.552 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - onSessionActive: NodeId{ns=1, id=Session:54229d6c-3ad5-411d-9df9-81619474f024}
12:31:52.725 [milo-shared-thread-pool-1] INFO  o.e.m.e.client.SubscriptionExample - item created for nodeId=NodeId{ns=0, id=2258}
12:31:53.580 [milo-shared-thread-pool-1] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338611125910000, javaDate=Wed May 13 12:31:52 EDT 2020}}
12:31:54.649 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338611135920000, javaDate=Wed May 13 12:31:53 EDT 2020}}
12:31:55.672 [milo-shared-thread-pool-0] INFO  o.e.m.e.client.SubscriptionExample - subscription value received: item=NodeId{ns=0, id=2258}, value=Variant{value=DateTime{utcTime=132338611145930000, javaDate=Wed May 13 12:31:54 EDT 2020}}