navcore_signalk_client/

stream.rs

//! Keeping the data stream open.
//!
//! A blocking connection on its own thread rather than an asynchronous
//! one. The clients this feeds are user interfaces with main loops of
//! their own, and asking each of them to host an async runtime for one
//! socket is a poor trade. What comes out is a plain
//! channel of events, which any main loop can drain.
//!
//! The reconnection is the point of the whole file. A wireless link on a
//! boat drops; a client that needed restarting afterwards would be
//! useless. Nothing here reports failure upwards as fatal -- it reports
//! that the link is down, keeps trying, and lets the consumer decide what
//! to show meanwhile. That decision has already been made elsewhere: the
//! fix goes stale and says so.
//!
//! # Being told when to look, without being told how
//!
//! `wake`, passed to [`Connection::open`], is called from this thread the
//! moment [`Connection::drain`] has something worth calling for. It is
//! deliberately the only thing this crate knows about whichever main loop
//! is on the other end -- a plain `Fn() + Send`, not a toolkit-specific
//! idle source or anything else naming a toolkit, so that a client on
//! any platform, and `sk-probe` on a bare terminal, all satisfy it the
//! same way each already satisfies its own main loop.
//!
//! It is *told*, not *handed the data*: [`Event`]s still only ever leave
//! through [`Connection::drain`], on the consumer's own timing. A callback
//! invoked once per message would turn a burst of deltas after a reconnect
//! into a burst of redraws, which is the one thing polling a receiver on a
//! loop protected against by accident; `wake` protects against it on
//! purpose, firing once for a burst and staying quiet until the consumer
//! has drained and something new arrives after that.

use std::io;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

use signalk::Delta;
use tungstenite::client::IntoClientRequest;
use tungstenite::http::HeaderValue;

use crate::{ClientError, Trust};

/// How long to wait before trying again, and the ceiling it grows to.
///
/// Starts short because most drops are a moment of interference; grows so
/// that a server which is genuinely off does not get hammered all night.
const FIRST_RETRY: Duration = Duration::from_secs(1);
const LONGEST_RETRY: Duration = Duration::from_secs(30);

/// How long [`read_until_broken`]'s own read blocks for before checking
/// [`Connection::publish`]'s own outgoing queue and trying again.
///
/// Short enough that a message [`Connection::publish`] just queued reaches
/// the wire promptly; long enough that this is not a busy-loop -- the same
/// order of magnitude [`crate::access`]'s own polling already uses for "a
/// human is somewhere in this loop, not a machine."
const READ_POLL_INTERVAL: Duration = Duration::from_millis(250);

/// What the connection reports.
#[derive(Debug, Clone, PartialEq)]
pub enum Event {
    /// The stream is open. Carries the vessel identifier the server states
    /// in its greeting, which is what own-ship deltas arrive under -- in
    /// full, not as "vessels.self".
    Connected {
        /// The identifier from the greeting, if it had one.
        self_id: Option<String>,
    },
    /// One message, already read into this workspace's units.
    Delta(Delta),
    /// The link is down. The consumer keeps whatever it last had, and lets
    /// it go stale on screen.
    Disconnected {
        /// What went wrong, for the log rather than for the screen.
        reason: String,
    },
    /// The server refused the handshake with `401 Unauthorized` -- the
    /// token this connection was opened with is no longer accepted,
    /// whether because it expired or because the administrator revoked
    /// the device (the two look identical from here: nothing in a 401
    /// itself says which). Told apart from an ordinary
    /// [`Event::Disconnected`] because the right response is different in
    /// kind, not just in wording: discard the token and ask again, rather
    /// than retry the same one.
    Unauthorized,
}

/// A running connection.
///
/// Dropping it asks the thread to stop after its current wait.
#[derive(Debug)]
pub struct Connection {
    events: Receiver<Event>,
    /// Shared with the background thread, which sets it -- see [`Sink`] --
    /// and cleared here, in [`Connection::drain`], on the consumer's own
    /// schedule rather than the thread's.
    pending: Arc<AtomicBool>,
    stop: Arc<AtomicBool>,
    /// Messages queued for [`read_until_broken`]'s own thread to write to
    /// the socket -- see [`Connection::publish`].
    outgoing: Sender<String>,
}

impl Connection {
    /// Opens the stream and keeps it open, reconnecting for as long as the
    /// connection lives.
    ///
    /// `token` is optional, but a server with security on and `allowReadonly`
    /// off will open this stream, greet the client, and then send it nothing
    /// -- see [`crate::access`]. Passing `None` is therefore a choice to be
    /// made knowingly, not a default to fall back on.
    ///
    /// `trust` decides which server may answer a `wss://` URL. It is asked
    /// for on every connection rather than defaulted, because the failure it
    /// prevents is silent: a client that trusted anything would reconnect
    /// happily to whatever took the boat's address and hand it the token.
    ///
    /// `wake` is called from this connection's own thread once something
    /// is worth draining -- see the module doc. A consumer with nothing
    /// better to do than poll on a fixed timer may pass `|| {}` and get
    /// exactly today's behaviour; passing the real thing is what lets the
    /// timer go away.
    #[must_use]
    pub fn open(
        stream_url: &str,
        token: Option<String>,
        trust: &Trust,
        wake: impl Fn() + Send + 'static,
    ) -> Self {
        let (sender, events) = channel();
        let stop = Arc::new(AtomicBool::new(false));
        let pending = Arc::new(AtomicBool::new(false));
        let sink = Sink {
            sender,
            pending: Arc::clone(&pending),
            wake: Box::new(wake),
        };
        let (outgoing, outgoing_events) = channel();

        let url = stream_url.to_owned();
        let stopper = Arc::clone(&stop);
        let trust = trust.clone();
        thread::spawn(move || run(&url, token.as_deref(), &trust, &sink, &stopper, &outgoing_events));

        Self {
            events,
            pending,
            stop,
            outgoing,
        }
    }

    /// Queues `delta_json` to be sent on the wire, as-is -- building a
    /// well-formed delta is the caller's own job (see
    /// [`signalk::notification::mob_delta`] for the one this exists for
    /// today). One-way: an ordinary delta carries no reply for this to
    /// wait on, unlike [`crate::resources::Client`]'s own writes.
    ///
    /// # Errors
    ///
    /// If the background thread this would be sent from is already gone
    /// -- the connection was dropped, or never opened.
    pub fn publish(&self, delta_json: String) -> Result<(), ClientError> {
        self.outgoing
            .send(delta_json)
            .map_err(|_| ClientError::Http("the connection is no longer running".to_owned()))
    }

    /// Whatever has arrived, without waiting.
    ///
    /// Drained on the consumer's own schedule rather than handed over
    /// message by message: the consumer decides when it is ready to draw,
    /// which is what keeps a burst of deltas from turning into a burst of
    /// redraws. `wake` (see the module doc) is what tells it *when* that
    /// schedule should run sooner than usual; this is still the only way
    /// the events themselves are read.
    ///
    /// Marked not-pending before the drain, not after: a message arriving
    /// *during* this call must still cause a future wake, and clearing
    /// first means the worst case is one redundant wake, never a missed
    /// one.
    pub fn drain(&self) -> impl Iterator<Item = Event> + '_ {
        self.pending.store(false, Ordering::Relaxed);
        self.events.try_iter()
    }
}

impl Drop for Connection {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

/// Where an [`Event`] goes, and the one place that decides whether that
/// was worth a `wake`.
///
/// A thin wrapper around the channel rather than calling `wake` at every
/// `send` site by hand -- which would be easy to get right once and then
/// forget at the next call site this file grows.
struct Sink {
    sender: Sender<Event>,
    pending: Arc<AtomicBool>,
    wake: Box<dyn Fn() + Send>,
}

impl Sink {
    /// Sends one event, waking the consumer only on the transition from
    /// nothing pending to something pending -- once per burst, exactly the
    /// property the module doc promises.
    fn send(&self, event: Event) -> Result<(), ()> {
        self.sender.send(event).map_err(|_| ())?;
        if !self.pending.swap(true, Ordering::Relaxed) {
            (self.wake)();
        }
        Ok(())
    }
}

/// The thread: connect, read until it breaks, wait, connect again.
fn run(url: &str, token: Option<&str>, trust: &Trust, sink: &Sink, stop: &AtomicBool, outgoing: &Receiver<String>) {
    let mut retry = FIRST_RETRY;

    while !stop.load(Ordering::Relaxed) {
        match connect(url, token, trust) {
            Ok(mut socket) => {
                retry = FIRST_RETRY;
                let reason = read_until_broken(&mut socket, sink, stop, outgoing);
                if sink.send(Event::Disconnected { reason }).is_err() {
                    return;
                }
            }
            Err(ConnectFailure::Unauthorized) => {
                // Not worth this loop's own back-off: the socket was
                // refused instantly, not lost to a flaky link, and
                // retrying the same token on a timer would just refuse
                // instantly again until the consumer -- the only side
                // that can discard a token and ask for a new one -- acts
                // on this event. Reported once and left to them.
                if sink.send(Event::Unauthorized).is_err() {
                    return;
                }
                return;
            }
            Err(ConnectFailure::Other(reason)) => {
                if sink.send(Event::Disconnected { reason }).is_err() {
                    return;
                }
            }
        }

        // Slept in short pieces so that dropping the connection is noticed
        // promptly rather than after half a minute.
        let mut waited = Duration::ZERO;
        while waited < retry && !stop.load(Ordering::Relaxed) {
            thread::sleep(Duration::from_millis(100));
            waited += Duration::from_millis(100);
        }
        retry = (retry * 2).min(LONGEST_RETRY);
    }
}

/// Why [`connect`] could not open a socket.
///
/// A plain `String` cannot be reacted to as anything but a message for
/// the log -- and a 401 is not a message for the log, it is an
/// instruction:
/// the token is no longer any good, discard it and ask again. Keeping
/// that one case structured, rather than grepping the rendered string
/// for "401" afterwards, is what lets [`run`] tell a consumer which of
/// the two it is looking at.
enum ConnectFailure {
    /// The handshake was refused with `401 Unauthorized`.
    Unauthorized,
    /// Anything else -- a network failure, a TLS failure, a different
    /// HTTP status, a handshake tungstenite could not make sense of.
    Other(String),
}

impl From<String> for ConnectFailure {
    fn from(reason: String) -> Self {
        Self::Other(reason)
    }
}

/// One connection attempt.
fn connect(
    url: &str,
    token: Option<&str>,
    trust: &Trust,
) -> Result<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>, ConnectFailure>
{
    // Subscribing at connect time rather than sending a subscription
    // afterwards: it is one round trip fewer, and it means a reconnect
    // needs no state to be replayed. "all", not "self": other vessels --
    // AIS targets -- are read this same way now (see a client's own
    // delta-feed handling for its Other-context branch), so this
    // asks for every context in one subscription rather than a second
    // one; `signalk::delta::parse`'s own "skip what is not recognised"
    // discipline is what makes the extra traffic for paths this crate
    // does not read at all (some other vessel's own wind data, say)
    // harmless rather than a growing list this has to filter by hand.
    let separator = if url.contains('?') { '&' } else { '?' };
    let mut request = format!("{url}{separator}subscribe=all")
        .into_client_request()
        .map_err(|error| error.to_string())?;

    if let Some(token) = token {
        let value =
            HeaderValue::from_str(&format!("Bearer {token}")).map_err(|error| error.to_string())?;
        request.headers_mut().insert("Authorization", value);
    }

    // The connector is built from the boat's authority rather than left to
    // tungstenite's default, which would be the public roots -- and no
    // public authority can ever vouch for a server called something.local.
    let connector = trust
        .rustls_config()
        .map(tungstenite::Connector::Rustls)
        .map_err(|error| error.to_string())?;

    let socket = dial(request.uri()).map_err(ConnectFailure::Other)?;
    // Set before the TLS handshake, not after: Rustls reads through to
    // this same underlying socket for every read it ever does, TLS
    // handshake included, so the timeout has to be in place from the
    // first byte -- see read_until_broken's own doc for what it is for.
    socket
        .set_read_timeout(Some(READ_POLL_INTERVAL))
        .map_err(|error| ConnectFailure::Other(error.to_string()))?;

    tungstenite::client_tls_with_config(request, socket, None, Some(connector))
        .map(|(socket, _)| socket)
        .map_err(|error| match &error {
            // The one case worth telling apart -- see ConnectFailure's own
            // doc. tungstenite hands back the server's real response
            // here, so this reads its actual status rather than assuming
            // 401 is the only reason a handshake gets refused at all.
            // `Interrupted` never applies to a blocking handshake like
            // this one and has no status to read, so it falls through to
            // `Other` alongside every other failure.
            tungstenite::HandshakeError::Failure(tungstenite::Error::Http(response))
                if response.status() == tungstenite::http::StatusCode::UNAUTHORIZED =>
            {
                ConnectFailure::Unauthorized
            }
            _ => ConnectFailure::Other(error.to_string()),
        })
}

/// The plain socket underneath, dialled from the request's own URI.
///
/// Taken from the parsed URI rather than by cutting up the URL text: the
/// port is optional, and the host of an IPv6 URL is bracketed there and
/// must not be when it reaches the resolver.
fn dial(uri: &tungstenite::http::Uri) -> Result<std::net::TcpStream, String> {
    let host = uri.host().ok_or_else(|| "no host in the URL".to_owned())?;
    let host = host.trim_start_matches('[').trim_end_matches(']');

    let port = uri.port_u16().unwrap_or(
        match tungstenite::client::uri_mode(uri) {
            Ok(tungstenite::stream::Mode::Tls) => 443,
            _ => 80,
        },
    );

    std::net::TcpStream::connect((host, port)).map_err(|error| format!("{host}:{port}: {error}"))
}

/// Reads messages until the socket fails, reporting what it understood.
fn read_until_broken(
    socket: &mut tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>,
    sink: &Sink,
    stop: &AtomicBool,
    outgoing: &Receiver<String>,
) -> String {
    // Learned from the greeting and then handed to every parse. Without
    // it own-ship deltas -- which arrive under the vessel's full
    // identifier, never under "vessels.self" -- read as another vessel,
    // and a consumer filtering for own ship sees nothing at all.
    let mut self_id: Option<String> = None;
    let mut greeted = false;

    while !stop.load(Ordering::Relaxed) {
        let text = match socket.read() {
            Ok(tungstenite::Message::Text(text)) => text.to_string(),
            Ok(tungstenite::Message::Close(_)) => return "server closed the stream".to_owned(),
            // Ping, pong and binary frames are not ours to interpret.
            Ok(_) => continue,
            // The read timeout set in connect() firing, not a real
            // failure -- see READ_POLL_INTERVAL's own doc. This is the
            // one moment this loop is not blocked inside socket.read(),
            // so it is also the only place Connection::publish's own
            // queue can actually reach the wire.
            Err(ref error) if is_read_timeout(error) => {
                match send_outgoing(socket, outgoing) {
                    Ok(()) => continue,
                    Err(reason) => return reason,
                }
            }
            Err(error) => return error.to_string(),
        };

        // The greeting comes first and names the vessel. It is not a delta,
        // which is why the reader has to know about it at all.
        if !greeted {
            greeted = true;
            self_id = greeting_self(&text);
            if sink
                .send(Event::Connected {
                    self_id: self_id.clone(),
                })
                .is_err()
            {
                return "nobody is listening any more".to_owned();
            }
            if self_id.is_some() {
                continue;
            }
        }

        if let Some(delta) = signalk::parse(&text, self_id.as_deref()) {
            if sink.send(Event::Delta(delta)).is_err() {
                return "nobody is listening any more".to_owned();
            }
        }
    }

    "asked to stop".to_owned()
}

/// Whether `error` is [`connect`]'s own read timeout firing, per
/// [`READ_POLL_INTERVAL`], rather than a real socket failure -- pulled out
/// as a plain function so this one judgement call, the whole reason
/// [`Connection::publish`] can ever reach the wire, is testable without a
/// real socket.
fn is_read_timeout(error: &tungstenite::Error) -> bool {
    matches!(
        error,
        tungstenite::Error::Io(io_error)
            if matches!(io_error.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut)
    )
}

/// Writes whatever [`Connection::publish`] has queued since the last time
/// this ran, in order, stopping at the first one that fails to send --
/// the same "something is wrong with this socket" signal a broken read
/// already is, reported the same way rather than as some third kind of
/// failure this crate would have to teach a caller to also react to.
/// Called only from [`read_until_broken`]'s own read-timeout branch,
/// never mid-frame: `socket.read()`'s own internal buffering means this
/// is safe exactly when a read attempt has just cleanly timed out with
/// nothing partial in flight.
fn send_outgoing(
    socket: &mut tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>,
    outgoing: &Receiver<String>,
) -> Result<(), String> {
    while let Ok(text) = outgoing.try_recv() {
        socket.send(tungstenite::Message::text(text)).map_err(|error| error.to_string())?;
    }
    Ok(())
}

/// The vessel identifier out of a server greeting, if that is what this is.
fn greeting_self(text: &str) -> Option<String> {
    let value: serde_json::Value = serde_json::from_str(text).ok()?;
    // A greeting names the server and the vessel and carries no updates.
    if value.get("updates").is_some() {
        return None;
    }
    value
        .get("self")
        .and_then(serde_json::Value::as_str)
        .map(|id| id.trim_start_matches("vessels.").to_owned())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_would_block_error_is_read_as_the_poll_timeout() {
        let error = tungstenite::Error::Io(io::Error::from(io::ErrorKind::WouldBlock));
        assert!(is_read_timeout(&error));
    }

    #[test]
    fn a_timed_out_error_is_read_as_the_poll_timeout_too() {
        // Some platforms report a blocking-socket read timeout as
        // TimedOut rather than WouldBlock; both mean the same thing here.
        let error = tungstenite::Error::Io(io::Error::from(io::ErrorKind::TimedOut));
        assert!(is_read_timeout(&error));
    }

    #[test]
    fn a_connection_reset_is_not_a_poll_timeout() {
        let error = tungstenite::Error::Io(io::Error::from(io::ErrorKind::ConnectionReset));
        assert!(!is_read_timeout(&error));
    }

    #[test]
    fn a_non_io_error_is_not_a_poll_timeout() {
        assert!(!is_read_timeout(&tungstenite::Error::AlreadyClosed));
    }

    #[test]
    fn publishing_after_the_connection_thread_is_gone_is_an_error() {
        let (outgoing, receiver) = channel();
        drop(receiver);
        let connection = Connection {
            events: channel().1,
            pending: Arc::new(AtomicBool::new(false)),
            stop: Arc::new(AtomicBool::new(false)),
            outgoing,
        };
        assert!(connection.publish("{}".to_owned()).is_err());
    }

    #[test]
    fn the_greeting_names_the_vessel() {
        // What a real server sends first, with the identifier prefixed by
        // "vessels." -- which the delta context also carries, so it is
        // stripped once here rather than compared for at every message.
        let self_id = greeting_self(
            r#"{"name":"signalk-server","version":"2.27.0",
                "self":"vessels.urn:mrn:signalk:uuid:0000","roles":["master","main"]}"#,
        );
        assert_eq!(self_id.as_deref(), Some("urn:mrn:signalk:uuid:0000"));
    }

    #[test]
    fn a_delta_is_not_mistaken_for_a_greeting() {
        assert!(
            greeting_self(
                r#"{"context":"vessels.urn:mrn:signalk:uuid:0000","updates":[
                   {"$source":"n2k.1","values":[
                   {"path":"navigation.speedOverGround","value":1.0}]}]}"#
            )
            .is_none()
        );
    }

    #[test]
    fn a_greeting_without_a_vessel_is_still_a_greeting() {
        assert!(greeting_self(r#"{"name":"signalk-server","version":"2.27.0"}"#).is_none());
    }

    fn sink() -> (Sink, Receiver<Event>, Arc<AtomicBool>) {
        let (sender, events) = channel();
        let pending = Arc::new(AtomicBool::new(false));
        let woken = Arc::new(AtomicBool::new(false));
        let sink = Sink {
            sender,
            pending: Arc::clone(&pending),
            wake: {
                let woken = Arc::clone(&woken);
                Box::new(move || woken.store(true, Ordering::Relaxed))
            },
        };
        (sink, events, woken)
    }

    #[test]
    fn the_first_message_of_a_burst_wakes_the_consumer() {
        let (sink, _events, woken) = sink();
        assert!(!woken.load(Ordering::Relaxed));
        sink.send(Event::Disconnected {
            reason: "test".to_owned(),
        })
        .expect("channel is open");
        assert!(woken.load(Ordering::Relaxed));
    }

    #[test]
    fn a_second_message_before_the_drain_wakes_nobody_again() {
        // The property the module doc promises: a burst wakes the consumer
        // once, not once per message, so draining stays the only thing
        // that decides when a redraw happens.
        let (sink, _events, woken) = sink();
        sink.send(Event::Disconnected {
            reason: "one".to_owned(),
        })
        .unwrap();
        woken.store(false, Ordering::Relaxed);

        sink.send(Event::Disconnected {
            reason: "two".to_owned(),
        })
        .unwrap();
        assert!(!woken.load(Ordering::Relaxed), "a second wake in the same burst");
    }

    #[test]
    fn draining_clears_pending_so_the_next_burst_can_wake_again() {
        let (sender, events) = channel();
        sender
            .send(Event::Disconnected {
                reason: "queued before drain".to_owned(),
            })
            .unwrap();

        let connection = Connection {
            events,
            pending: Arc::new(AtomicBool::new(true)),
            stop: Arc::new(AtomicBool::new(false)),
            outgoing: channel().0,
        };

        assert_eq!(connection.drain().count(), 1);
        // What the module doc's promise actually rests on: with pending
        // left set, Sink::send's swap would see it already true and never
        // call wake again, no matter how long the link then stayed quiet.
        assert!(!connection.pending.load(Ordering::Relaxed));
    }
}