surreal_client/
live.rs

1//! Live-query notifications.
2//!
3//! SurrealDB pushes a change frame over the same WebSocket for every row a
4//! `LIVE SELECT` matches. Those frames are *not* replies to a request — they
5//! arrive unsolicited, tagged with the live-query id the server returned from
6//! the `live` RPC. The engine's read loop demultiplexes them from ordinary
7//! responses and forwards each as a [`Notification`] onto a channel that backs
8//! a [`LiveStream`].
9
10use std::pin::Pin;
11use std::task::{Context, Poll};
12
13use ciborium::Value as CborValue;
14use futures::Stream;
15use tokio::sync::mpsc;
16
17/// What happened to a row a live query is watching.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum Action {
20    /// A row entered the watched set.
21    Create,
22    /// A watched row's contents changed.
23    Update,
24    /// A row left the watched set.
25    Delete,
26}
27
28impl Action {
29    /// Parse SurrealDB's `CREATE` / `UPDATE` / `DELETE` action string
30    /// (case-insensitive). Returns `None` for anything else (e.g. `KILLED`).
31    pub fn parse(s: &str) -> Option<Self> {
32        match s.to_ascii_uppercase().as_str() {
33            "CREATE" => Some(Action::Create),
34            "UPDATE" => Some(Action::Update),
35            "DELETE" => Some(Action::Delete),
36            _ => None,
37        }
38    }
39}
40
41/// One live-query change frame: which query, what happened, and the record.
42///
43/// `data` is the raw CBOR record exactly as SurrealDB delivered it (a map of
44/// fields including `id`). Consumers decode it with the same CBOR→value path
45/// they use for ordinary reads. `record_id` is the affected row's id (a
46/// SurrealDB `Thing`), present on every action — including `Delete`, where
47/// `data` may be empty.
48#[derive(Debug, Clone)]
49pub struct Notification {
50    /// The live-query id this frame belongs to (matches [`LiveStream::query_id`]).
51    pub query_id: String,
52    /// The kind of change.
53    pub action: Action,
54    /// The affected row's id, as raw CBOR (a `Thing`).
55    pub record_id: CborValue,
56    /// The affected record as raw CBOR (empty/`Null` on some `Delete`s).
57    pub data: CborValue,
58}
59
60/// A stream of [`Notification`]s for one live query.
61///
62/// Yields until the underlying WebSocket closes. Dropping the stream stops
63/// local delivery but does **not** send `KILL` to the server — call
64/// [`SurrealClient::kill`](crate::SurrealClient::kill) with
65/// [`query_id`](Self::query_id) to release the server-side query.
66pub struct LiveStream {
67    pub(crate) query_id: String,
68    pub(crate) rx: mpsc::UnboundedReceiver<Notification>,
69}
70
71impl LiveStream {
72    /// The server-assigned live-query id (a UUID string).
73    pub fn query_id(&self) -> &str {
74        &self.query_id
75    }
76
77    /// Await the next notification, or `None` once the connection closes.
78    pub async fn recv(&mut self) -> Option<Notification> {
79        self.rx.recv().await
80    }
81}
82
83impl Stream for LiveStream {
84    type Item = Notification;
85
86    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
87        self.rx.poll_recv(cx)
88    }
89}
surreal_client/live.rs

surreal_client/
live.rs
