Emit "connected" event (#27)

README.md

@@ -76,6 +76,10 @@ For details see [dist/index.d.ts](./dist/index.d.ts).
 
 ## Error & event handling
 
+#### `instance.events.on("connected", listener: () => void)`
+
+The `connected` event is emitted once after initially establishing the connection and later once after every successful reconnect. Reconnects happen automatically when `pg-listen` detects that the connection closed or became unresponsive.
+
 #### `instance.events.on("error", listener: (error: Error) => void)`
 
 An `error` event is emitted for fatal errors that affect the notification subscription. A standard way of handling those kinds of errors would be to `console.error()`-log the error and terminate the process with a non-zero exit code.

dist/index.d.ts

@@ -6,6 +6,7 @@ export interface PgParsedNotification {
     payload?: any;
 }
 interface PgListenEvents {
+    connected: () => void;
     error: (error: Error) => void;
     notification: (notification: PgParsedNotification) => void;
     reconnect: (attempt: number) => void;

src/index.ts

@@ -26,6 +26,7 @@ export interface PgParsedNotification {
 }
 
 interface PgListenEvents {
+  connected: () => void,
   error: (error: Error) => void,
   notification: (notification: PgParsedNotification) => void,
   reconnect: (attempt: number) => void
@@ -78,7 +79,7 @@ export interface Options {
   serialize?: (data: any) => string
 }
 
-function connect (connectionConfig: pg.ClientConfig | undefined, options: Options) {
+function connect(connectionConfig: pg.ClientConfig | undefined, emitter: TypedEventEmitter<PgListenEvents>, options: Options) {
   connectionLogger("Creating PostgreSQL client for notification streaming")
 
   const { retryInterval = 500, retryLimit = Infinity, retryTimeout = 3000 } = options
@@ -203,7 +204,7 @@ function createPostgresSubscriber (connectionConfig?: pg.ClientConfig, options:
     notificationsEmitter.emit(notification.channel, notification.payload)
   })
 
-  const { dbClient: initialDBClient, reconnect } = connect(connectionConfig, options)
+  const { dbClient: initialDBClient, reconnect } = connect(connectionConfig, emitter, options)
 
   let closing = false
   let dbClient = initialDBClient
@@ -257,6 +258,8 @@ function createPostgresSubscriber (connectionConfig?: pg.ClientConfig, options:
       await Promise.all(subscribedChannels.map(
         channelName => dbClient.query(`LISTEN ${format.ident(channelName)}`)
       ))
+
+      emitter.emit("connected")
     } catch (error) {
       error.message = `Re-initializing the PostgreSQL notification client after connection loss failed: ${error.message}`
       connectionLogger(error.stack || error)
@@ -276,9 +279,10 @@ function createPostgresSubscriber (connectionConfig?: pg.ClientConfig, options:
     notifications: notificationsEmitter,
 
     /** Don't forget to call this asyncronous method before doing your thing */
-    connect () {
+    async connect () {
       initialize(dbClient)
-      return dbClient.connect()
+      await dbClient.connect()
+      emitter.emit("connected")
     },
     close () {
       connectionLogger("Closing PostgreSQL notification listener.")

test/integration.test.ts

@@ -16,16 +16,20 @@ test("can connect", async t => {
 })
 
 test("can listen and notify", async t => {
+  let connectedEvents = 0
   const notifications: PgParsedNotification[] = []
   const receivedPayloads: any[] = []
 
   const hub = createPostgresSubscriber({ connectionString: "postgres://postgres:postgres@localhost:5432/postgres" })
+
+  hub.events.on("connected", () => connectedEvents++)
+  hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
+  hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
+
   await hub.connect()
 
   try {
     await hub.listenTo("test")
-    await hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
-    await hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
 
     await hub.notify("test", { hello: "world" })
     await hub.notify("test2", "should not be received, since not subscribed to channel test2")
@@ -42,6 +46,7 @@ test("can listen and notify", async t => {
     t.deepEqual(receivedPayloads, [
       { hello: "world" }
     ])
+    t.is(connectedEvents, 1)
   } finally {
     await hub.close()
   }
@@ -56,8 +61,9 @@ test("can handle notification without payload", async t => {
 
   try {
     await hub.listenTo("test")
-    await hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
-    await hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
+
+    hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
+    hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
 
     await hub.notify("test")
     await delay(200)
@@ -110,9 +116,11 @@ test("can use custom `parse` function", async t => {
 })
 
 test.serial("getting notified after connection is terminated", async t => {
+  let connectedEvents = 0
+  let reconnects = 0
+
   const notifications: PgParsedNotification[] = []
   const receivedPayloads: any[] = []
-  let reconnects = 0
 
   const connectionString = "postgres://postgres:postgres@localhost:5432/postgres"
   let client = new pg.Client({ connectionString })
@@ -122,13 +130,16 @@ test.serial("getting notified after connection is terminated", async t => {
     { connectionString: connectionString + "?ApplicationName=pg-listen-termination-test" },
     { paranoidChecking: 1000 }
   )
+
+  hub.events.on("connected", () => connectedEvents++)
+  hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
+  hub.events.on("reconnect", () => reconnects++)
+  hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
+
   await hub.connect()
 
   try {
     await hub.listenTo("test")
-    hub.events.on("notification", (notification: PgParsedNotification) => notifications.push(notification))
-    hub.events.on("reconnect", () => reconnects++)
-    hub.notifications.on("test", (payload: any) => receivedPayloads.push(payload))
 
     await delay(1000)
     debug("Terminating database backend")
@@ -156,6 +167,7 @@ test.serial("getting notified after connection is terminated", async t => {
       { hello: "world" }
     ])
     t.is(reconnects, 1)
+    t.is(connectedEvents, 2)
   } finally {
     debug("Closing the subscriber")
     await hub.close()