replace listen example with a small chat example

Co-authored-by: Stephen <webmaster@scd31.com>

Cargo.toml

@@ -14,7 +14,7 @@ members = [
     "examples/postgres/axum-social-with-tests",
     "examples/postgres/files",
     "examples/postgres/json",
-    "examples/postgres/listen",
+    "examples/postgres/chat",
     "examples/postgres/todos",
     "examples/postgres/mockable-todos",
     "examples/postgres/transaction",

examples/postgres/chat/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "sqlx-example-postgres-chat"
+version = "0.1.0"
+edition = "2021"
+workspace = "../../../"
+
+[dependencies]
+sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio-native-tls" ] }
+futures = "0.3.1"
+tokio = { version = "1.20.0", features = [ "rt-multi-thread", "macros" ] }
+tui = "0.19.0"
+crossterm = "0.25"
+unicode-width = "0.1"

examples/postgres/chat/README.md

@@ -0,0 +1,21 @@
+# Chat Example
+
+## Description
+
+This example demonstrates how to use PostgreSQL channels to create a very simple chat application.
+
+## Setup
+
+1. Declare the database URL
+
+    ```
+    export DATABASE_URL="postgres://postgres:password@localhost/files"
+    ```
+
+## Usage
+
+Run the project
+
+```
+cargo run -p sqlx-examples-postgres-chat
+```

examples/postgres/chat/src/main.rs

@@ -0,0 +1,177 @@
+use crossterm::{
+    event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode},
+    execute,
+    terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
+};
+use sqlx::postgres::PgListener;
+use sqlx::PgPool;
+use std::sync::Arc;
+use std::{error::Error, io};
+use tokio::{sync::Mutex, time::Duration};
+use tui::{
+    backend::{Backend, CrosstermBackend},
+    layout::{Constraint, Direction, Layout},
+    style::{Color, Modifier, Style},
+    text::{Span, Spans, Text},
+    widgets::{Block, Borders, List, ListItem, Paragraph},
+    Frame, Terminal,
+};
+use unicode_width::UnicodeWidthStr;
+
+struct ChatApp {
+    input: String,
+    messages: Arc<Mutex<Vec<String>>>,
+    pool: PgPool,
+}
+
+impl ChatApp {
+    fn new(pool: PgPool) -> Self {
+        ChatApp {
+            input: String::new(),
+            messages: Arc::new(Mutex::new(Vec::new())),
+            pool,
+        }
+    }
+
+    async fn run<B: Backend>(
+        mut self,
+        terminal: &mut Terminal<B>,
+        mut listener: PgListener,
+    ) -> Result<(), Box<dyn Error>> {
+        // setup listener task
+        let messages = self.messages.clone();
+        std::mem::drop(tokio::spawn(async move {
+            while let Ok(msg) = listener.recv().await {
+                messages.lock().await.push(msg.payload().to_string());
+            }
+        }));
+
+        loop {
+            let messages: Vec<ListItem> = self
+                .messages
+                .lock()
+                .await
+                .iter()
+                .map(|m| {
+                    let content = vec![Spans::from(Span::raw(m.to_owned()))];
+                    ListItem::new(content)
+                })
+                .collect();
+
+            terminal.draw(|f| self.ui(f, messages))?;
+
+            if !event::poll(Duration::from_millis(20))? {
+                continue;
+            }
+
+            if let Event::Key(key) = event::read()? {
+                match key.code {
+                    KeyCode::Enter => {
+                        notify(&self.pool, self.input.drain(..).collect()).await?;
+                    }
+                    KeyCode::Char(c) => {
+                        self.input.push(c);
+                    }
+                    KeyCode::Backspace => {
+                        self.input.pop();
+                    }
+                    KeyCode::Esc => {
+                        return Ok(());
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+
+    fn ui<B: Backend>(&mut self, f: &mut Frame<B>, messages: Vec<ListItem>) {
+        let chunks = Layout::default()
+            .direction(Direction::Vertical)
+            .margin(2)
+            .constraints(
+                [
+                    Constraint::Length(1),
+                    Constraint::Length(3),
+                    Constraint::Min(1),
+                ]
+                .as_ref(),
+            )
+            .split(f.size());
+
+        let text = Text::from(Spans::from(vec![
+            Span::raw("Press "),
+            Span::styled("Enter", Style::default().add_modifier(Modifier::BOLD)),
+            Span::raw(" to send the message, "),
+            Span::styled("Esc", Style::default().add_modifier(Modifier::BOLD)),
+            Span::raw(" to quit"),
+        ]));
+        let help_message = Paragraph::new(text);
+        f.render_widget(help_message, chunks[0]);
+
+        let input = Paragraph::new(self.input.as_ref())
+            .style(Style::default().fg(Color::Yellow))
+            .block(Block::default().borders(Borders::ALL).title("Input"));
+        f.render_widget(input, chunks[1]);
+        f.set_cursor(
+            // Put cursor past the end of the input text
+            chunks[1].x + self.input.width() as u16 + 1,
+            // Move one line down, from the border to the input line
+            chunks[1].y + 1,
+        );
+
+        let messages =
+            List::new(messages).block(Block::default().borders(Borders::ALL).title("Messages"));
+        f.render_widget(messages, chunks[2]);
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // setup postgres
+    let conn_str =
+        std::env::var("DATABASE_URL").expect("Env var DATABASE_URL is required for this example.");
+    let pool = sqlx::PgPool::connect(&conn_str).await?;
+
+    let mut listener = PgListener::connect(&conn_str).await?;
+    listener.listen_all(vec!["chan0"]).await?;
+
+    // setup terminal
+    enable_raw_mode()?;
+    let mut stdout = io::stdout();
+    execute!(stdout, EnterAlternateScreen, EnableMouseCapture)?;
+    let backend = CrosstermBackend::new(stdout);
+    let mut terminal = Terminal::new(backend)?;
+
+    // create app and run it
+    let app = ChatApp::new(pool);
+    let res = app.run(&mut terminal, listener).await;
+
+    // restore terminal
+    disable_raw_mode()?;
+    execute!(
+        terminal.backend_mut(),
+        LeaveAlternateScreen,
+        DisableMouseCapture,
+    )?;
+    terminal.show_cursor()?;
+
+    if let Err(err) = res {
+        println!("{:?}", err)
+    }
+
+    Ok(())
+}
+
+async fn notify(pool: &PgPool, s: String) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        r#"
+SELECT pg_notify(chan, payload)
+FROM (VALUES ('chan0', $1)) v(chan, payload)
+"#,
+    )
+    .bind(s)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}