demonstrate alternative constructor w/ listener support

Add a `NewWithListener` constructor to `riverdatabasesql` that allows
the `database/sql` driver to be used with a functioning listener
implementation. Also add a `NewListener` constructor to the `riverpgxv5`
driver to allow creating a listener with a raw pgx pool.

These can be combined to allow full listener support as long as the
underlying database driver supports it, even when it's used within an
abstraction like `database/sql` or Bun.

Author: bgentry

riverdriver/riverdatabasesql/river_database_sql.go

@@ -24,8 +24,9 @@ import (
 
 // Driver is an implementation of riverdriver.Driver for database/sql.
 type Driver struct {
-	dbPool  *sql.DB
-	queries *dbsqlc.Queries
+	dbPool   *sql.DB
+	listener riverdriver.Listener
+	queries  *dbsqlc.Queries
 }
 
 // New returns a new database/sql River driver for use with River.
@@ -41,13 +42,28 @@ func New(dbPool *sql.DB) *Driver {
 	return &Driver{dbPool: dbPool, queries: dbsqlc.New()}
 }
 
+// NewWithListener returns a new database/sql River driver for use with River
+// just like New, except it also takes a riverdriver.Listener to use for
+// listening to notifications.
+func NewWithListener(dbPool *sql.DB, listener riverdriver.Listener) *Driver {
+	driver := New(dbPool)
+	driver.listener = listener
+	return driver
+}
+
 func (d *Driver) GetExecutor() riverdriver.Executor {
 	return &Executor{d.dbPool, d.dbPool, dbsqlc.New()}
 }
 
-func (d *Driver) GetListener() riverdriver.Listener { panic(riverdriver.ErrNotImplemented) }
-func (d *Driver) HasPool() bool                     { return d.dbPool != nil }
-func (d *Driver) SupportsListener() bool            { return false }
+func (d *Driver) GetListener() riverdriver.Listener {
+	if d.listener == nil {
+		panic(riverdriver.ErrNotImplemented)
+	}
+	return d.listener
+}
+
+func (d *Driver) HasPool() bool          { return d.dbPool != nil }
+func (d *Driver) SupportsListener() bool { return d.listener != nil }
 
 func (d *Driver) UnwrapExecutor(tx *sql.Tx) riverdriver.ExecutorTx {
 	return &ExecutorTx{Executor: Executor{nil, tx, dbsqlc.New()}, tx: tx}

riverdriver/riverpgxv5/river_pgx_v5_driver.go

@@ -567,6 +567,26 @@ type Listener struct {
 	mu     sync.Mutex
 }
 
+// NewListener returns a Listener that can be used to enable other drivers
+// (notably `riverdatabasesql`) to listen for notifications when their
+// abstraction doesn't allow direct access to pgx connections, even though they
+// are using pgx under the hood.
+//
+// Users of `database/sql` or Bun can use this with the
+// `riverdatabasesql.NewWithListener` constructor to enable listener support in
+// that driver.
+//
+// The dbPool will solely be used for acquiring new connections for `LISTEN`
+// commands. As such, a pool must be provided that supports that command. Users
+// of pgbouncer should ensure that this specific pool is configured with session
+// pooling, even if their main application does not use session pooling.
+//
+// A single Client will never use more than one listener connection at a time,
+// no matter how many topics are being listened to.
+func NewListener(dbPool *pgxpool.Pool) riverdriver.Listener {
+	return &Listener{dbPool: dbPool}
+}
+
 func (l *Listener) Close(ctx context.Context) error {
 	l.mu.Lock()
 	defer l.mu.Unlock()