LISTEN/NOTIFY 与 COPY

pub struct PgListener {
    pool: Pool<Postgres>,
    connection: Option<PoolConnection<Postgres>>,
    buffer_rx: mpsc::UnboundedReceiver<Notification>,
    buffer_tx: Option<mpsc::UnboundedSender<Notification>>,
    channels: Vec<String>,
    ignore_close_event: bool,
    eager_reconnect: bool,
}

        // Create a pool of 1 without timeouts (as they don't apply here)
        // We only use the pool to handle re-connections
        let pool = PoolOptions::<Postgres>::new()
            .max_connections(1)
            .max_lifetime(None)
            .idle_timeout(None)
            .connect(url)
            .await?;

let mut listener = PgListener::connect_with(&pool).await?; // pool: max 1, 无 lifetime / idle 超时
listener.listen("pg2_ch").await?;
let pid: i32 = sqlx::query_scalar("select pg_backend_pid()").fetch_one(&mut listener).await?;

notifier.execute("notify pg2_ch, 'one'").await?;
let n = listener.recv().await?;
println!("recv {:?} from pid {}", n.payload(), n.process_id());

// 通知在 listener 自己跑查询时到达
listener.execute("notify pg2_ch, 'self'").await?;
println!("next_buffered {:?}", listener.next_buffered().map(|n| n.payload().to_string()));

  [conn 1] C->S Query("LISTEN \"pg2_ch\"")
  [conn 1] S->C CommandComplete("LISTEN"), ReadyForQuery(I)
  ...
listener pid 393
--- 1. 普通的一条 ---
  [conn 1] S->C NotificationResponse
recv "one" from pid 392
--- 2. 通知在 listener 自己跑查询时到达 ---
  [conn 1] C->S Query("notify pg2_ch, 'self'")
  [conn 1] S->C CommandComplete("NOTIFY"), NotificationResponse, ReadyForQuery(I)
next_buffered Some("self")

            let message = match res {
                Ok(message) => message,

                // The connection is dead, ensure that it is dropped,
                // update self state, and loop to try again.
                Err(Error::Io(err))
                    if matches!(
                        err.kind(),
                        io::ErrorKind::ConnectionAborted |
                        io::ErrorKind::UnexpectedEof |
                        // see ERRORS section in tcp(7) man page (https://man7.org/linux/man-pages/man7/tcp.7.html)
                        io::ErrorKind::TimedOut |
                        io::ErrorKind::BrokenPipe
                    ) =>
                {
                    if let Some(mut conn) = self.connection.take() {
                        self.buffer_tx = conn.inner.stream.notifications.take();
                        // Close the connection in a background task, so we can continue.
                        conn.close_on_drop();
                    }

                    if self.eager_reconnect {
                        self.connect_if_needed().await?;
                    }

                    // lost connection
                    return Ok(None);
                }

notifier.execute("notify pg2_ch, 'before-kill'").await?;
let killed: bool = sqlx::query_scalar("select pg_terminate_backend($1)").bind(pid).fetch_one(&mut notifier).await?;
tokio::time::sleep(Duration::from_millis(200)).await;
notifier.execute("notify pg2_ch, 'while-down'").await?;
tokio::time::sleep(Duration::from_millis(200)).await;
for _ in 0..2 {
  let r = listener.try_recv().await?;
  println!("try_recv {:?}", r.map(|n| n.payload().to_string()));
}
notifier.execute("notify pg2_ch, 'after'").await?;
let r = listener.try_recv().await?;
println!("try_recv {:?}", r.map(|n| n.payload().to_string()));

--- 3. 杀掉 listener 自己的后端 ---
  [conn 1] S->C NotificationResponse
pg_terminate_backend(393) = true
  [conn 1] S->C ErrorResponse
try_recv Some("before-kill")
  [conn 1] C->S Terminate
  [conn 2] C->S StartupMessage(v3.0)
  ...
  [conn 2] C->S Query("LISTEN \"pg2_ch\";")
  [conn 2] S->C CommandComplete("LISTEN"), ReadyForQuery(I)
try_recv None
  [conn 2] S->C NotificationResponse
try_recv Some("after")
  ...
listener pid 394

    async fn begin(mut conn: C, statement: &str) -> Result<Self> {
        conn.wait_until_ready().await?;
        conn.inner.stream.send(Query(statement)).await?;

        let response = match conn.inner.stream.recv_expect::<CopyInResponse>().await {
            Ok(res) => res.0,
            Err(e) => {
                conn.inner.stream.recv().await?;
                return Err(e);
            }
        };

        Ok(PgCopyIn {
            conn: Some(conn),
            response,
        })
    }

let mut copy = conn.copy_in_raw("copy item from stdin (format csv)").await?;
println!("is_textual {} num_columns {}", copy.is_textual(), copy.num_columns());
copy.send(b"1,apple\n2,banana\n".as_slice()).await?;
copy.send(b"3,cherry\n".as_slice()).await?;
let rows = copy.finish().await?;

let chunks: Vec<_> = conn.copy_out_raw("copy item to stdout (format csv)").await?.try_collect().await?;

let mut copy = conn.copy_in_raw("copy item from stdin (format csv)").await?;
copy.send(b"4,durian\n".as_slice()).await?;
copy.abort("changed my mind").await?;

--- COPY FROM STDIN ---
  [conn 1] C->S Query("copy item from stdin (format csv)")
  [conn 1] S->C CopyInResponse
is_textual true num_columns 2
  [conn 1] C->S CopyData(17 字节)
  [conn 1] C->S CopyData(9 字节), CopyDone
  [conn 1] S->C CommandComplete("COPY 3"), ReadyForQuery(I)
finish -> 3 行
--- COPY TO STDOUT ---
  [conn 1] C->S Query("copy item to stdout (format csv)")
  [conn 1] S->C CopyOutResponse, CopyData(8 字节), CopyData(9 字节), CopyData(9 字节), CopyDone, CommandComplete("COPY 3"), ReadyForQuery(I)
chunk "1,apple\n"
chunk "2,banana\n"
chunk "3,cherry\n"
--- abort ---
  [conn 1] C->S Query("copy item from stdin (format csv)")
  [conn 1] S->C CopyInResponse
  [conn 1] C->S CopyData(9 字节)
  [conn 1] C->S CopyFail
  [conn 1] S->C ErrorResponse
  [conn 1] S->C ReadyForQuery(I)
abort -> Ok

impl<C: DerefMut<Target = PgConnection>> Drop for PgCopyIn<C> {
    fn drop(&mut self) {
        if let Some(mut conn) = self.conn.take() {
            conn.inner
                .stream
                .write_msg(CopyFail::new(
                    "PgCopyIn dropped without calling finish() or fail()",
                ))
                .expect("BUG: PgCopyIn abort message should not be too large");
        }
    }
}

key BigInt(5367643390023655937)
--- A acquire ---
  [conn 1] C->S Parse(sqlx_s_1, "SELECT pg_advisory_lock($1)"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1; [8] 4a 7d b3 39 67 55 62 01), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 0), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
B try_acquire -> 没拿到
--- drop(guard) ---
B try_acquire -> 没拿到
--- A 再用一次 ---
  [conn 1] C->S Query("SELECT pg_advisory_unlock(5367643390023655937)")
  [conn 1] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
  [conn 1] C->S Query("select 1")
  [conn 1] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
B try_acquire -> 拿到

--- 既不 finish 也不 abort，直接 drop ---
  [conn 1] C->S Query("copy item from stdin (format csv)")
  [conn 1] S->C CopyInResponse
  [conn 1] C->S CopyData(13 字节)
  [conn 1] C->S CopyFail
  [conn 1] C->S Parse(sqlx_s_1, "select $1 * 100"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ErrorResponse
  [conn 1] S->C ReadyForQuery(I)
drop 之后第 1 次查询 -> Database(Some("57014"), COPY from stdin failed: PgCopyIn dropped without calling finish() or fail())
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_2, "select $1 * 100"), Describe(S, sqlx_s_2), Sync
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2; [8] 00 00 00 00 00 00 00 02), Execute, Close(P, unnamed), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
drop 之后第 2 次查询 -> RowNotFound
  [conn 1] S->C BindComplete, DataRow(列长 8), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
drop 之后第 3 次查询 -> Ok(200)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2; [8] 00 00 00 00 00 00 00 03), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 8), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
drop 之后第 4 次查询 -> Ok(300)