spg_engine/
notify.rs

1//! v7.39 (round 222) — LISTEN / NOTIFY delivery. Was accept-and-drop
2//! since v7.37.17; now notifications are real: LISTEN subscribes the
3//! session, NOTIFY queues on the transaction and delivers at COMMIT
4//! (PG semantics: transactional, deduplicated within the tx, dropped
5//! at ROLLBACK; immediate under autocommit), and the wire layer drains
6//! [`Engine::take_notifications`] into 'A' NotificationResponse
7//! messages after each statement. Delivery is same-process (the
8//! engine's session-state architecture wall, like `session_params`):
9//! every pgwire connection shares the engine, so a NOTIFY from one
10//! connection reaches a LISTEN from another at the next statement
11//! boundary. Idle-connection push (no statement in flight) is not
12//! implemented — psycopg2 / libpq poll patterns receive on their next
13//! interaction.
14
15use alloc::string::String;
16use alloc::vec::Vec;
17
18use crate::{EngineError, QueryResult};
19
20impl crate::Engine {
21    /// Drain committed notifications (channel, payload). Wire layers emit
22    /// each as a NotificationResponse; embedded callers consume directly.
23    pub fn take_notifications(&mut self) -> Vec<(String, String)> {
24        core::mem::take(&mut self.delivered_notifies)
25    }
26
27    /// COMMIT boundary — release the transaction's pending notifications
28    /// to the delivery queue (only channels someone LISTENs on).
29    pub(crate) fn notifies_on_commit(&mut self) {
30        let pending = core::mem::take(&mut self.tx_pending_notifies);
31        for (ch, payload) in pending {
32            if self.listen_channels.contains(&ch) {
33                self.delivered_notifies.push((ch, payload));
34            }
35        }
36    }
37
38    /// ROLLBACK boundary — the aborted transaction's notifications vanish.
39    pub(crate) fn notifies_on_rollback(&mut self) {
40        self.tx_pending_notifies.clear();
41    }
42
43    pub(crate) fn exec_listen(&mut self, channel: String) -> Result<QueryResult, EngineError> {
44        self.listen_channels.insert(channel);
45        Ok(QueryResult::CommandOk {
46            affected: 0,
47            modified_catalog: false,
48        })
49    }
50
51    pub(crate) fn exec_unlisten(
52        &mut self,
53        channel: Option<String>,
54    ) -> Result<QueryResult, EngineError> {
55        match channel {
56            Some(c) => {
57                self.listen_channels.remove(&c);
58            }
59            None => self.listen_channels.clear(),
60        }
61        Ok(QueryResult::CommandOk {
62            affected: 0,
63            modified_catalog: false,
64        })
65    }
66
67    pub(crate) fn exec_notify(
68        &mut self,
69        channel: String,
70        payload: Option<String>,
71    ) -> Result<QueryResult, EngineError> {
72        let payload = payload.unwrap_or_default();
73        if self.in_transaction() {
74            // PG deduplicates identical (channel, payload) pairs within one
75            // transaction.
76            if !self
77                .tx_pending_notifies
78                .iter()
79                .any(|(c, p)| *c == channel && *p == payload)
80            {
81                self.tx_pending_notifies.push((channel, payload));
82            }
83        } else if self.listen_channels.contains(&channel) {
84            // Autocommit: immediate delivery.
85            self.delivered_notifies.push((channel, payload));
86        }
87        Ok(QueryResult::CommandOk {
88            affected: 0,
89            modified_catalog: false,
90        })
91    }
92}
spg_engine/notify.rs

spg_engine/
notify.rs
