Real-Time Database Notifications
with PostgreSQL LISTEN/NOTIFY

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install psycopg2-binary python-dotenv

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.7.2</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.17.0</version>
</dependency>

go mod init listen-notify
go get github.com/jackc/pgx/v5
go get github.com/joho/godotenv

cursor.execute("LISTEN table_changes;")
cursor.execute("LISTEN user_events;")
cursor.execute("LISTEN payments;")

# notify.channel tells you which channel fired
while True:
    if select.select([conn], [], [], 5.0) == ([conn], [], []):
        conn.poll()
        while conn.notifies:
            notify = conn.notifies.pop(0)
            if notify.channel == 'table_changes':
                pass  # handle table changes
            elif notify.channel == 'payments':
                pass  # handle payments

Statement stmt = conn.createStatement();
stmt.execute("LISTEN table_changes");
stmt.execute("LISTEN user_events");
stmt.execute("LISTEN payments");

// note.getName() tells you which channel fired
PGNotification[] notes = pgConn.getNotifications(500);
if (notes != null) {
    for (PGNotification note : notes) {
        if ("table_changes".equals(note.getName())) { /* ... */ }
        if ("payments".equals(note.getName()))      { /* ... */ }
    }
}

conn.Exec(ctx, "LISTEN table_changes")
conn.Exec(ctx, "LISTEN user_events")
conn.Exec(ctx, "LISTEN payments")

// note.Channel tells you which channel fired
for {
    note, _ := conn.WaitForNotification(ctx)
    switch note.Channel {
    case "table_changes":
        // handle table changes
    case "payments":
        // handle payments
    }
}

import time, os, select, psycopg2
import psycopg2.extensions

def start_listener_with_backoff():
    retry_delay = 1.0  # start at 1 second

    while True:
        try:
            conn = psycopg2.connect(os.getenv('DATABASE_URL'))
            conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
            retry_delay = 1.0  # reset on success

            cursor = conn.cursor()
            cursor.execute("LISTEN table_changes;")
            print("Connected and listening...")

            while True:
                if select.select([conn], [], [], 5.0) == ([conn], [], []):
                    conn.poll()
                    while conn.notifies:
                        handle_notification(conn.notifies.pop(0))

        except Exception as e:
            print(f"Lost ({e}). Reconnecting in {retry_delay}s...")
            time.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, 30.0)  # exponential backoff max 30s

public static void listenWithBackoff() {
    long retryDelay = 1000; // start at 1 second

    while (true) {
        try (Connection conn = DriverManager.getConnection(System.getenv("DATABASE_URL"))) {
            retryDelay = 1000; // reset on success
            Statement stmt = conn.createStatement();
            stmt.execute("LISTEN table_changes");
            PGConnection pgConn = conn.unwrap(PGConnection.class);

            System.out.println("Connected and listening...");
            while (true) {
                PGNotification[] notes = pgConn.getNotifications(500);
                if (notes != null) {
                    for (PGNotification note : notes) handleNotification(note);
                }
            }
        } catch (Exception e) {
            System.err.println("Lost. Reconnecting in " + (retryDelay/1000) + "s...");
            try { Thread.sleep(retryDelay); } catch (Exception ie) {}
            retryDelay = Math.min(retryDelay * 2, 30000); // exponential backoff max 30s
        }
    }
}

func listenWithBackoff() {
    retryDelay := 1 * time.Second

    for {
        ctx := context.Background()
        conn, err := pgx.Connect(ctx, os.Getenv("DATABASE_URL"))
        if err != nil {
            log.Printf("Connect failed: %v. Reconnecting in %v...", err, retryDelay)
            time.Sleep(retryDelay)
            if retryDelay < 30*time.Second {
                retryDelay *= 2
            }
            continue
        }

        retryDelay = 1 * time.Second // reset on success
        conn.Exec(ctx, "LISTEN table_changes")
        log.Println("Connected and listening...")

        for {
            note, err := conn.WaitForNotification(ctx)
            if err != nil {
                log.Printf("Connection lost: %v", err)
                conn.Close(ctx)
                break
            }
            handleNotification(note)
        }
        time.Sleep(retryDelay)
        if retryDelay < 30*time.Second {
            retryDelay *= 2
        }
    }
}