opcua_client/session/
connect.rs

1use std::sync::Arc;
2
3use tokio::{pin, select};
4use tracing::{info, info_span, Instrument};
5
6use crate::transport::{Connector, SecureChannelEventLoop, TransportPollResult};
7use opcua_types::{Error, NodeId};
8
9use super::Session;
10
11/// This struct manages the task of connecting to the server.
12/// It will only make a single attempt, so whatever is calling it is responsible for retries.
13pub(super) struct SessionConnector {
14    inner: Arc<Session>,
15}
16
17/// When the session connects to the server, this describes
18/// how that happened, whether a new session was created, or an old session was reactivated.
19#[derive(Debug, Clone)]
20pub enum SessionConnectMode {
21    /// A new session was created with session ID given by the inner [`NodeId`]
22    NewSession(NodeId),
23    /// An old session was reactivated with session ID given by the inner [`NodeId`]
24    ReactivatedSession(NodeId),
25}
26
27impl SessionConnector {
28    pub(super) fn new(session: Arc<Session>) -> Self {
29        Self { inner: session }
30    }
31
32    pub(super) async fn try_connect<T: Connector>(
33        &self,
34        connector: &T,
35    ) -> Result<(SecureChannelEventLoop<T::Transport>, SessionConnectMode), Error> {
36        self.connect_and_activate(connector).await
37    }
38
39    async fn connect_and_activate<T: Connector>(
40        &self,
41        connector: &T,
42    ) -> Result<(SecureChannelEventLoop<T::Transport>, SessionConnectMode), Error> {
43        let span = info_span!(
44            "Attempting to create and activate session to server",
45            endpoint_url = connector.default_endpoint().endpoint_url.as_ref()
46        );
47        let mut event_loop = self
48            .inner
49            .channel
50            .connect_no_retry(connector)
51            .instrument(span.clone())
52            .await?;
53
54        let activate_fut = self.ensure_and_activate_session().instrument(span.clone());
55        pin!(activate_fut);
56
57        let res = loop {
58            select! {
59                r = event_loop.poll() => {
60                    if let TransportPollResult::Closed(c) = r {
61                        return Err(Error::new(c, "Transport closed while activating session"));
62                    }
63                },
64                r = &mut activate_fut => break r,
65            }
66        };
67
68        let id = match res {
69            Ok(id) => id,
70            Err(e) => {
71                self.inner.channel.close_channel().instrument(span).await;
72
73                loop {
74                    if matches!(event_loop.poll().await, TransportPollResult::Closed(_)) {
75                        break;
76                    }
77                }
78
79                return Err(e);
80            }
81        };
82
83        Ok((event_loop, id))
84    }
85
86    async fn ensure_and_activate_session(&self) -> Result<SessionConnectMode, Error> {
87        let should_create_session = self.inner.session_id.load().is_null();
88
89        if should_create_session {
90            self.inner.create_session().in_current_span().await?;
91        }
92
93        let reconnect = match self.inner.activate_session().in_current_span().await {
94            Err(status_code) if !should_create_session => {
95                info!(
96                    "Session activation failed on reconnect, error = {}, creating a new session",
97                    status_code
98                );
99                self.inner.reset();
100                let id = self.inner.create_session().in_current_span().await?;
101                self.inner.activate_session().in_current_span().await?;
102                SessionConnectMode::NewSession(id)
103            }
104            Err(e) => return Err(e),
105            Ok(_) => {
106                let session_id = (**self.inner.session_id.load()).clone();
107                if should_create_session {
108                    SessionConnectMode::NewSession(session_id)
109                } else {
110                    SessionConnectMode::ReactivatedSession(session_id)
111                }
112            }
113        };
114
115        if self.inner.recreate_subscriptions {
116            self.inner
117                .transfer_subscriptions_from_old_session()
118                .in_current_span()
119                .await;
120        }
121
122        Ok(reconnect)
123    }
124}
opcua_client/session/connect.rs

opcua_client/session/
connect.rs
