LISTEN/NOTIFY funcionality

Signed-off-by: chandr-andr (Kiselev Aleksandr) <chandr@chandr.net>

Author: chandr-andr

docs/.vuepress/sidebar.ts

@@ -23,6 +23,7 @@ export default sidebar({
         "connection",
         "transaction",
         "cursor",
+        "listener",
         "results",
         "exceptions",
       ],

docs/components/connection_pool.md

@@ -254,6 +254,18 @@ This is the preferable way to work with the PostgreSQL.
 :::
 
 
+### Listener
+
+Create a new instance of a listener.
+
+```python
+async def main() -> None:
+    ...
+    listener = db_pool.listener()
+```
+```
+
+
 ### Close
 
 To close the connection pool at the stop of your application.

docs/components/listener.md

@@ -0,0 +1,204 @@
+---
+title: Listener
+---
+
+`Listener` object allows users to work with [LISTEN](https://www.postgresql.org/docs/current/sql-listen.html)/[NOTIFY](https://www.postgresql.org/docs/current/sql-notify.html) functionality.
+
+## Usage
+
+There two ways of using `Listener` object:
+- Async iterator
+- Background task
+
+::: tabs
+
+@tab Background task
+```python
+from psqlpy import ConnectionPool, Connection, Listener
+
+
+db_pool = ConnectionPool(
+    dsn="postgres://postgres:postgres@localhost:5432/postgres",
+)
+
+async def test_channel_callback(
+    connection: Connection,
+    payload: str,
+    channel: str,
+    process_id: int,
+) -> None:
+    # do some important staff
+    ...
+
+async def main() -> None:
+    # Create listener object
+    listener: Listener = db_pool.listener()
+
+    # Add channel to listen and callback for it.
+    await listener.add_callback(
+        channel="test_channel",
+        callback=test_channel_callback,
+    )
+
+    # Startup the listener
+    await listener.startup()
+
+    # Start listening.
+    # `listen` method isn't blocking, it returns None and starts background
+    # task in the Rust event loop.
+    listener.listen()
+
+    # You can stop listening.
+    listener.abort_listen()
+```
+
+@tab Async Iterator
+```python
+from psqlpy import (
+    ConnectionPool,
+    Connection,
+    Listener,
+    ListenerNotificationMsg,
+)
+
+
+db_pool = ConnectionPool(
+    dsn="postgres://postgres:postgres@localhost:5432/postgres",
+)
+
+async def main() -> None:
+    # Create listener object
+    listener: Listener = db_pool.listener()
+
+    # Startup the listener
+    await listener.startup()
+
+    listener_msg: ListenerNotificationMsg
+    async for listener_msg in listener:
+        print(listener_msg)
+```
+
+:::
+
+## Listener attributes
+
+- `connection`: Instance of `Connection`.
+If `startup` wasn't called, raises `ListenerStartError`.
+
+- `is_started`: Flag that shows whether the `Listener` is running or not.
+
+## Listener methods
+
+### Startup
+
+Startup `Listener` instance and can be called once or again only after `shutdown`.
+
+::: important
+`Listener` must be started up.
+:::
+
+```python
+async def main() -> None:
+    listener: Listener = db_pool.listener()
+
+    await listener.startup()
+```
+
+### Shutdown
+Abort listen (if called) and release underlying connection.
+
+```python
+async def main() -> None:
+    listener: Listener = db_pool.listener()
+
+    await listener.startup()
+    await listener.shutdown()
+```
+
+### Add Callback
+
+#### Parameters:
+- `channel`: name of the channel to listen.
+- `callback`: coroutine callback.
+
+Add new callback to the channel, can be called more than 1 times.
+
+Callback signature is like this:
+```python
+from psqlpy import Connection
+
+async def callback(
+    connection: Connection,
+    payload: str,
+    channel: str,
+    process_id: int,
+) -> None:
+    ...
+```
+
+Parameters for callback are based like `args`, so this signature is correct to:
+```python
+async def callback(
+    connection: Connection,
+    *args,
+) -> None:
+    ...
+```
+
+**Example:**
+```python
+async def test_channel_callback(
+    connection: Connection,
+    payload: str,
+    channel: str,
+    process_id: int,
+) -> None:
+    ...
+
+async def main() -> None:
+    listener = db_pool.listener()
+
+    await listener.add_callback(
+        channel="test_channel",
+        callback=test_channel_callback,
+    )
+```
+
+### Clear Channel Callbacks
+
+#### Parameters:
+- `channel`: name of the channel
+
+Remove all callbacks for the channel
+
+```python
+async def main() -> None:
+    listener = db_pool.listener()
+    await listener.clear_channel_callbacks()
+```
+
+### Clear All Channels
+Clear all channels and callbacks.
+
+```python
+async def main() -> None:
+    listener = db_pool.listener()
+    await listener.clear_all_channels()
+```
+
+### Listen
+Start listening.
+
+It's a non-blocking operation.
+In the background it creates task in Rust event loop.
+
+```python
+async def main() -> None:
+    listener = db_pool.listener()
+    await listener.startup()
+    await listener.listen()
+```
+
+### Abort Listen
+Abort listen.
+If `listen()` method was called, stop listening, else don't do anything.

python/psqlpy/_internal/__init__.pyi

@@ -1809,6 +1809,7 @@ class Listener:
     """
 
     connection: Connection
+    is_started: bool
 
     def __aiter__(self: Self) -> Self: ...
     async def __anext__(self: Self) -> ListenerNotificationMsg: ...
@@ -1824,6 +1825,11 @@ class Listener:
 
         Each listener MUST be started up.
         """
+    async def shutdown(self: Self) -> None:
+        """Shutdown the listener.
+
+        Abort listen and release underlying connection.
+        """
     async def add_callback(
         self: Self,
         channel: str,
@@ -1844,7 +1850,7 @@ class Listener:
         ) -> None: ...
         ```
 
-        Callback parameters are passed as args.
+        Callback parameters are passed as args on the Rust side.
         """
 
     async def clear_channel_callbacks(self, channel: str) -> None:

src/driver/connection_pool.rs

@@ -127,7 +127,7 @@ pub fn connect(
     let mgr: Manager = build_manager(
         mgr_config,
         pg_config.clone(),
-        build_tls(&ca_file, ssl_mode)?,
+        build_tls(&ca_file, &ssl_mode)?,
     );
 
     let mut db_pool_builder = Pool::builder(mgr);

src/driver/connection_pool_builder.rs

@@ -53,7 +53,7 @@ impl ConnectionPoolBuilder {
         let mgr: Manager = build_manager(
             mgr_config,
             self.config.clone(),
-            build_tls(&self.ca_file, self.ssl_mode)?,
+            build_tls(&self.ca_file, &self.ssl_mode)?,
         );
 
         let mut db_pool_builder = Pool::builder(mgr);

src/driver/listener/core.rs

@@ -161,8 +161,19 @@ impl Listener {
     }
 
     #[getter]
-    fn connection(&self) -> Connection {
-        self.connection.clone()
+    fn is_started(&self) -> bool {
+        self.is_started
+    }
+
+    #[getter]
+    fn connection(&self) -> RustPSQLDriverPyResult<Connection> {
+        if !self.is_started {
+            return Err(RustPSQLDriverError::ListenerStartError(
+                "Listener isn't started up".into(),
+            ));
+        }
+
+        Ok(self.connection.clone())
     }
 
     async fn startup(&mut self) -> RustPSQLDriverPyResult<()> {
@@ -172,7 +183,7 @@ impl Listener {
             ));
         }
 
-        let tls_ = build_tls(&self.ca_file.clone(), self.ssl_mode)?;
+        let tls_ = build_tls(&self.ca_file, &self.ssl_mode)?;
 
         let mut builder = SslConnector::builder(SslMethod::tls())?;
         builder.set_verify(SslVerifyMode::NONE);
@@ -214,6 +225,14 @@ impl Listener {
         Ok(())
     }
 
+    async fn shutdown(&mut self) {
+        self.abort_listen();
+        std::mem::take(&mut self.connection);
+        std::mem::take(&mut self.receiver);
+
+        self.is_started = false;
+    }
+
     #[pyo3(signature = (channel, callback))]
     async fn add_callback(
         &mut self,

src/driver/utils.rs

@@ -181,7 +181,7 @@ pub enum ConfiguredTLS {
 /// May return Err Result if cannot create builder.
 pub fn build_tls(
     ca_file: &Option<String>,
-    ssl_mode: Option<SslMode>,
+    ssl_mode: &Option<SslMode>,
 ) -> RustPSQLDriverPyResult<ConfiguredTLS> {
     if let Some(ca_file) = ca_file {
         let mut builder = SslConnector::builder(SslMethod::tls())?;
@@ -190,7 +190,7 @@ pub fn build_tls(
             builder.build(),
         )));
     } else if let Some(ssl_mode) = ssl_mode {
-        if ssl_mode == common_options::SslMode::Require {
+        if *ssl_mode == common_options::SslMode::Require {
             let mut builder = SslConnector::builder(SslMethod::tls())?;
             builder.set_verify(SslVerifyMode::NONE);
             return Ok(ConfiguredTLS::TlsConnector(MakeTlsConnector::new(