java - milo 的 OPC 客户端无法连接到本地 OPC 发现服务

Question

我是 OPC UA 新手，我正在使用 milo OPC Subscriber client连接到本地发现服务。我有 Prosys 模拟服务器，它连接到我的本地发现服务。

注意:如果我直接连接到 prosys 端点，它工作正常。它仅通过发现服务失败。

运行代码时出现以下异常

<pre>12:38:35.916 [main] INFO org.eclipse.milo.opcua.stack.core.Stack -
Successfully removed cryptography restrictions. 12:38:36.167 [main]
INFO com.company.OpcuaClientRunner - security temp dir:
C:\Users\Z003Z2YP\AppData\Local\Temp\security 12:38:36.173 [main] INFO
com.company.KeyStoreLoader - Loading KeyStore at
C:\Users\Z003Z2YP\AppData\Local\Temp\security\example-client.pfx
12:38:37.594 [main] INFO com.company.OpcuaClientRunner - Using
endpoint: opc.tcp://<hostname>:4840 [None] 12:38:37.600 [main] INFO
org.eclipse.milo.opcua.sdk.client.OpcUaClient - Eclipse Milo OPC UA
Stack version: 0.2.3 12:38:37.600 [main] INFO
org.eclipse.milo.opcua.sdk.client.OpcUaClient - Eclipse Milo OPC UA
Client SDK version: 0.2.3 12:38:37.809 [NonceUtilSecureRandom] INFO
org.eclipse.milo.opcua.stack.core.util.NonceUtil - SecureRandom seeded
in 0ms. 12:38:37.815 [ua-netty-event-loop-1] ERROR
org.eclipse.milo.opcua.stack.client.handlers.UaTcpClientMessageHandler
- [remote=<hostname>/<IP>:4840] errorMessage=ErrorMessage{error=StatusCode{name=Bad_ServiceUnsupported,
value=0x800B0000, quality=bad}, reason=null} 12:38:53.828 [main] ERROR
com.company.OpcuaClientRunner - Error running client example:
UaException: status=Bad_Timeout, message=request timed out after
16000ms java.util.concurrent.ExecutionException: UaException:
status=Bad_Timeout, message=request timed out after 16000ms     
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
at com.company.OpcuaSubscriber.run(OpcuaSubscriber.java:49)     
at com.company.OpcuaClientRunner.run(OpcuaClientRunner.java:122)    
at com.company.OpcuaSubscriber.main(OpcuaSubscriber.java:120) Caused by:
org.eclipse.milo.opcua.stack.core.UaException: request timed out after
16000ms     
at org.eclipse.milo.opcua.stack.client.UaTcpStackClient.lambda$scheduleRequestTimeout$13(UaTcpStackClient.java:326)
at io.netty.util.HashedWheelTimer$HashedWheelTimeout.expire(HashedWheelTimer.java:581)
at io.netty.util.HashedWheelTimer$HashedWheelBucket.expireTimeouts(HashedWheelTimer.java:655)
at io.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:367)
at java.lang.Thread.run(Thread.java:748) 12:38:53.828 [main] ERROR
com.company.OpcuaClientRunner - Error running example: UaException:
status=Bad_Timeout, message=request timed out after 16000ms
java.util.concurrent.ExecutionException: UaException:
status=Bad_Timeout, message=request timed out after 16000ms     
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
at com.company.OpcuaSubscriber.run(OpcuaSubscriber.java:49)     
at com.company.OpcuaClientRunner.run(OpcuaClientRunner.java:122)    
at com.company.OpcuaSubscriber.main(OpcuaSubscriber.java:120) Caused by:
org.eclipse.milo.opcua.stack.core.UaException: request timed out after
16000ms     
at org.eclipse.milo.opcua.stack.client.UaTcpStackClient.lambda$scheduleRequestTimeout$13(UaTcpStackClient.java:326)
at io.netty.util.HashedWheelTimer$HashedWheelTimeout.expire(HashedWheelTimer.java:581)
at io.netty.util.HashedWheelTimer$HashedWheelBucket.expireTimeouts(HashedWheelTimer.java:655)
at io.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:367)
at java.lang.Thread.run(Thread.java:748)</pre>

在 ClientRunner 中创建客户端的代码。

private OpcUaClient createClient() throws Exception {        File securityTempDir = new File(System.getProperty("java.io.tmpdir"), "security");        if (!securityTempDir.exists() && !securityTempDir.mkdirs()) {            throw new Exception("unable to create security dir: " + securityTempDir);        }        LoggerFactory.getLogger(getClass())            .info("security temp dir: {}", securityTempDir.getAbsolutePath());        KeyStoreLoader loader = new KeyStoreLoader().load(securityTempDir);        loader.load();        SecurityPolicy securityPolicy = client.getSecurityPolicy();        EndpointDescription[] endpoints;        try {            endpoints = UaTcpStackClient                .getEndpoints(client.getEndpointUrl())                .get();        } catch (Throwable ex) {            ex.printStackTrace();            // try the explicit discovery endpoint as well            String discoveryUrl = client.getEndpointUrl();            logger.info("Trying explicit discovery URL: {}", discoveryUrl);            endpoints = UaTcpStackClient                .getEndpoints(discoveryUrl)                .get();        }        EndpointDescription endpoint = Arrays.stream(endpoints)            .filter(e -> e.getSecurityPolicyUri().equals(securityPolicy.getSecurityPolicyUri()))            .findFirst().orElseThrow(() -> new Exception("no desired endpoints returned"));        logger.info("Using endpoint: {} [{}]", endpoint.getEndpointUrl(), securityPolicy);        OpcUaClientConfig config = OpcUaClientConfig.builder()            .setApplicationName(LocalizedText.english("eclipse milo opc-ua client"))            .setApplicationUri("urn:eclipse:milo:examples:client")            .setCertificate(loader.getClientCertificate())            .setKeyPair(loader.getClientKeyPair())            .setEndpoint(endpoint)            .setIdentityProvider(client.getIdentityProvider())            .setRequestTimeout(uint(5000))            .build();        return new OpcUaClient(config);

客户端接口(interface)类

public interface OpcuaClientInterface {        public static final String USERNAME = "demo";        public static final String PASSWORD = "demo";        default String getEndpointUrl() {            return "opc.tcp://localhost:4840/UADiscovery";        }        default SecurityPolicy getSecurityPolicy() {             return SecurityPolicy.None;        }        default IdentityProvider getIdentityProvider() {    //       return new UsernameProvider(USERNAME,PASSWORD);            return new AnonymousProvider();        }        void run (OpcUaClient client, CompletableFuture future) throws Exception;    }

订阅者运行实现

    @Override    public void run(OpcUaClient client, CompletableFuture future) throws Exception {        // synchronous connect        client.connect().get();        // create a subscription @ 1000ms        UaSubscription subscription = client.getSubscriptionManager().createSubscription(1000.0).get();        List nodeIds = Arrays.asList("SuctionPressure", "DischargePressure", "Flow", "BearingTemperature", "Vibration", "Power");//      List nodeIds = Arrays.asList("DS", "PV");        List MICRs = nodeIds.stream().map(id -> {            ReadValueId readValueId = new ReadValueId(new NodeId(3, id), AttributeId.Value.uid(), null,                    QualifiedName.NULL_VALUE);            // important: client handle must be unique per item            UInteger clientHandle = uint(clientHandles.getAndIncrement());            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);            return request;        }).collect(Collectors.toList());        // 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 onItemCreated = (item, id) -> item                .setValueConsumer(this::onSubscriptionValue);        List items = subscription.createMonitoredItems(TimestampsToReturn.Both, MICRs, 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());            }        }        // let the example run for 5 seconds then terminate        // Thread.sleep(1000 * 60 * 1);        // future.complete(client);    }

Answer 1

我让它与milo OPC Subscriber client一起工作.

以下是我在类里面所做的更改。

客户端界面

public interface OpcuaClientInterface {

public static final String USERNAME = "demo";
public static final String PASSWORD = "demo";
public static final String UaServerName = "SimulationServer";

default String getEndpointUrl() {
    return "opc.tcp://localhost:4840";
}

default SecurityPolicy getSecurityPolicy() {
     return SecurityPolicy.None;
}

default String getUaServerName () {
    return UaServerName;
}


default IdentityProvider getIdentityProvider() {
     return new UsernameProvider(USERNAME,PASSWORD);
}

void run (OpcUaClient client, CompletableFuture<OpcUaClient> future) throws Exception;

}

客户端运行类

public class OpcuaClientRunner {

static {
    CryptoRestrictions.remove();
    Security.addProvider(new BouncyCastleProvider());
}
private final AtomicLong requestHandle = new AtomicLong(1L);

private final Logger logger = LoggerFactory.getLogger(getClass());
private final CompletableFuture<OpcUaClient> future = new CompletableFuture<>();

private final OpcuaClientInterface client;

public OpcuaClientRunner(OpcuaClientInterface client) throws Exception {
    this.client = client;
}

private KeyStoreLoader createKeyStore() {
    KeyStoreLoader loader = null;
    try {
        File securityTempDir = new File(System.getProperty("java.io.tmpdir"), "security");
        if (!securityTempDir.exists() && !securityTempDir.mkdirs()) {
            throw new Exception("unable to create security dir: " + securityTempDir);
        }
        LoggerFactory.getLogger(getClass()).info("security temp dir: {}", securityTempDir.getAbsolutePath());
        loader = new KeyStoreLoader().load(securityTempDir);
        loader.load();
    } catch (Exception e) {
        logger.error("Could not load keys {}", e);
        return null;
    }
    return loader;
}

private CompletableFuture<ServerOnNetwork[]> findServersOnNetwork(String discoveryEndpointUrl)
        throws InterruptedException, ExecutionException {
    UaStackClient c = createDiscoveryClient(client.getEndpointUrl()).connect().get();
    RequestHeader header = new RequestHeader(NodeId.NULL_VALUE, DateTime.now(),
            uint(requestHandle.getAndIncrement()), uint(0), null, uint(60), null);

    FindServersOnNetworkRequest request = new FindServersOnNetworkRequest(header, null, null, null);

    return c.<FindServersOnNetworkResponse>sendRequest(request).thenCompose(result -> {
        StatusCode statusCode = result.getResponseHeader().getServiceResult();

        if (statusCode.isGood()) {
            return CompletableFuture.completedFuture(result.getServers());
        } else {
            CompletableFuture<ServerOnNetwork[]> f = new CompletableFuture<>();
            f.completeExceptionally(new UaException(statusCode));
            return f;
        }
    });
}

private UaTcpStackClient createDiscoveryClient(String endpointUrl) {
    KeyStoreLoader loader = createKeyStore();
    if (loader == null) {
        return null;
    }
    UaTcpStackClientConfig config = UaTcpStackClientConfig.builder()
            .setApplicationName(LocalizedText.english("Stack Example Client"))
            .setApplicationUri(String.format("urn:example-client:%s", UUID.randomUUID()))
            .setCertificate(loader.getClientCertificate()).setKeyPair(loader.getClientKeyPair())
            .setEndpointUrl(endpointUrl).build();

    return new UaTcpStackClient(config);
}

private OpcUaClient createUaClient() throws Exception {

    KeyStoreLoader loader = createKeyStore();
    if (loader == null) {
        return null;
    }

    SecurityPolicy securityPolicy = client.getSecurityPolicy();
    EndpointDescription[] endpoints = null;

    try {
        ServerOnNetwork[] servers = findServersOnNetwork(client.getEndpointUrl()).get();
        ServerOnNetwork server = Arrays.stream(servers)
                .filter(e -> e.getServerName().equals(client.getUaServerName())).findFirst()
                .orElseThrow(() -> new Exception("no desired UA Server returned."));
        endpoints = UaTcpStackClient.getEndpoints(server.getDiscoveryUrl()).get();
    } catch (Throwable ex) {
        ex.printStackTrace();
    }

    EndpointDescription endpoint = Arrays.stream(endpoints)
            .filter(e -> e.getSecurityPolicyUri().equals(securityPolicy.getSecurityPolicyUri())).findFirst()
            .orElseThrow(() -> new Exception("no desired endpoints returned"));
    logger.info("Using endpoint: {} [{}]", endpoint.getEndpointUrl(), securityPolicy);

    OpcUaClientConfig config = OpcUaClientConfig.builder()
            .setApplicationName(LocalizedText.english("eclipse milo opc-ua client"))
            .setApplicationUri("urn:eclipse:milo:examples:client").setCertificate(loader.getClientCertificate())
            .setKeyPair(loader.getClientKeyPair()).setEndpoint(endpoint)
            .setIdentityProvider(client.getIdentityProvider()).setRequestTimeout(uint(5000)).build();

    return new OpcUaClient(config);
}

public void run() {
    try {
        OpcUaClient uaClient = createUaClient();
        future.whenComplete((c, ex) -> {
            if (ex != null) {
                logger.error("Error running example: {}", ex.getMessage(), ex);
            }

            try {
                uaClient.disconnect().get();
                Stack.releaseSharedResources();
            } catch (InterruptedException | ExecutionException e) {
                logger.error("Error disconnecting:", e.getMessage(), e);
            }

            try {
                Thread.sleep(1000);
                System.exit(0);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        try {
            client.run(uaClient, future);
            future.get(15, TimeUnit.SECONDS);
        } catch (Throwable t) {
            logger.error("Error running client example: {}", t.getMessage(), t);
            future.completeExceptionally(t);
        }

    } catch (Throwable t) {
        logger.error("Error getting client: {}", t.getMessage(), t);

        future.completeExceptionally(t);

        try {
            Thread.sleep(1000);
            System.exit(0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /*
     * try { Thread.sleep(999999999); } catch (InterruptedException e) {
     * e.printStackTrace(); }
     */
}

}