taktwerk 0.6.0

Fixed-step model execution engine for Linux with its own OPC UA server
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

//! An OPC UA client session to an engine's server, shared by `tui` and `record`.

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use opcua::client::{Client, ClientBuilder, IdentityToken, Session};
use opcua::crypto::SecurityPolicy;
use opcua::types::{MessageSecurityMode, StatusCode};
use tokio::task::JoinHandle;

/// How long connecting may take.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// A connected session.
pub struct Conn {
    _client: Client,
    /// The session.
    pub session: Arc<Session>,
    events: JoinHandle<StatusCode>,
}

impl Conn {
    /// Connect anonymously without security as application `app`; message limits fit a
    /// recorder's largest pull.
    pub async fn open(endpoint: &str, app: &str) -> Result<Self, String> {
        static N: AtomicU32 = AtomicU32::new(0);
        let pki = std::env::temp_dir().join(format!(
            "{app}-{}-{}",
            std::process::id(),
            N.fetch_add(1, Ordering::Relaxed)
        ));
        let client = ClientBuilder::new()
            .application_name(app)
            .application_uri(format!("urn:taktwerk:{app}"))
            .pki_dir(&pki)
            .create_sample_keypair(false)
            .trust_server_certs(true)
            .session_retry_limit(0)
            .max_message_size(64 << 20)
            .max_chunk_count(1024)
            .max_byte_string_length(16 << 20)
            .max_string_length(16 << 20)
            .client();
        let _ = std::fs::remove_dir_all(&pki);
        let mut client = client.map_err(|e| format!("client: {e:?}"))?;
        let (session, event_loop) = client
            .connect_to_matching_endpoint(
                (
                    endpoint,
                    SecurityPolicy::None.to_str(),
                    MessageSecurityMode::None,
                ),
                IdentityToken::Anonymous,
            )
            .await
            .map_err(|e| format!("connect {endpoint}: {e}"))?;
        let events = event_loop.spawn();
        tokio::time::timeout(CONNECT_TIMEOUT, session.wait_for_connection())
            .await
            .map_err(|_| format!("connect {endpoint}: timed out"))?;
        Ok(Self {
            _client: client,
            session,
            events,
        })
    }

    /// Whether the session's event loop has ended (the connection is gone for good).
    pub fn ended(&self) -> bool {
        self.events.is_finished()
    }

    /// Disconnect.
    pub async fn close(self) {
        let _ = self.session.disconnect().await;
        self.events.abort();
    }
}