Implement automatic reconnect for listen - fixes porsager#35

porsager · porsager · commit 5058a576c2ab · 2020-05-22T11:41:01.000+02:00
commit 003bc24f1f98af585a4276d787eeab8eafda980a
Author: Rasmus Porsager &lt;rasmus@porsager.com&gt;
Date:   Wed May 20 08:51:19 2020 +0200

    Clean up reconnect

commit 585c19491e0a8d3696833bfaf2bab24e24fc1f6c
Author: Rasmus Porsager &lt;rasmus@porsager.com&gt;
Date:   Mon Mar 9 20:19:32 2020 +0100

    Clean up tests

commit dc80681f4b91de6fdaab0b7576a55b106c8f463b
Author: Rasmus Porsager &lt;rasmus@porsager.com&gt;
Date:   Fri Feb 14 21:54:43 2020 +0100

    Reconnect listener connection on close
diff --git a/lib/backend.js b/lib/backend.js
@@ -83,6 +83,7 @@ function Backend({
         backend.query.result.count = +x.toString('utf8', i + 1, x.length - 1)
       if (x[i - 1] >= 65) {
         backend.query.result.command = x.toString('utf8', 5, i)
+        backend.query.result.state = state
         break
       }
     }
diff --git a/lib/connection.js b/lib/connection.js
@@ -18,6 +18,7 @@ function Connection(options = {}) {
     connect_timeout,
     onnotify,
     onnotice,
+    onclose,
     parsers
   } = options
   let buffer = Buffer.alloc(0)
@@ -230,6 +231,7 @@ function Connection(options = {}) {
     statements = {}
     messages = []
     open = ready = false
+    onclose && onclose()
   }
 
   /* c8 ignore next */
diff --git a/lib/index.js b/lib/index.js
@@ -340,16 +340,23 @@ function Postgres(a, b) {
     }
 
     listeners[channel] = [fn]
-    return query({ raw: true }, getListener(), 'listen ' + escape(channel)).then(() => channel)
+    return query({ raw: true }, getListener(), 'listen ' + escape(channel))
   }
 
   function getListener() {
     if (listener)
       return listener
 
-    listener = Connection(Object.assign({},
-      options,
-      { onnotify: (c, x) => c in listeners && listeners[c].forEach(fn => fn(x)) }
+    listener = Connection(Object.assign({
+      onnotify: (c, x) => c in listeners && listeners[c].forEach(fn => fn(x)),
+      onclose: () => {
+        Object.entries(listeners).forEach(([channel, fns]) => {
+          delete listeners[channel]
+          Promise.allSettled(fns.map(fn => listen(channel, fn)))
+        })
+      }
+    },
+      options
     ))
     all.push(listener)
     return listener
diff --git a/tests/index.js b/tests/index.js
@@ -5,6 +5,7 @@ const cp = require('child_process')
 const path = require('path')
 
 const postgres = require('../lib')
+const delay = ms => new Promise(r => setTimeout(r, ms))
 
 const login = {
   user: 'postgres_js_test'
@@ -30,7 +31,7 @@ const options = {
   user: login.user,
   pass: login.pass,
   idle_timeout: 0.2,
-  debug: true,
+  debug: false,
   max: 1
 }
 
@@ -385,7 +386,7 @@ t('sql file throws', async() =>
 
 t('sql file cached', async() => {
   await sql.file(path.join(__dirname, 'select.sql'))
-  await new Promise(r => setTimeout(r, 20))
+  await delay(20)
 
   return [1, (await sql.file(path.join(__dirname, 'select.sql')))[0].x]
 })
@@ -416,7 +417,6 @@ t('Connection ended timeout', async() => {
 
 t('Connection ended error', async() => {
   const sql = postgres(options)
-
   sql.end()
   return ['CONNECTION_ENDED', (await sql``.catch(x => x.code))]
 })
@@ -549,6 +549,21 @@ t('listen and notify with weird name', async() => {
   )]
 })
 
+t('listen reconnects', async() => {
+  const listener = postgres(options)
+      , xs = []
+
+  const { state: { pid } } = await listener.listen('test', x => xs.push(x))
+  await sql.notify('test', 'a')
+  await sql`select pg_terminate_backend(${ pid }::int)`
+  await delay(50)
+  await sql.notify('test', 'b')
+  await delay(50)
+  listener.end()
+
+  return ['ab', xs.join('')]
+})
+
 t('responds with server parameters (application_name)', async() =>
   ['postgres.js', await new Promise((resolve, reject) => postgres({
     ...options,
@@ -788,7 +803,7 @@ t('Cursor works', async() => {
   const order = []
   await sql`select 1 as x union select 2 as x`.cursor(async(x) => {
     order.push(x.x + 'a')
-    await new Promise(r => setTimeout(r, 100))
+    await delay(100)
     order.push(x.x + 'b')
   })
   return ['1a1b2a2b', order.join('')]
@@ -815,7 +830,7 @@ t('Cursor throw works', async() => {
   const order = []
   await sql`select 1 as x union select 2 as x`.cursor(async(x) => {
     order.push(x.x + 'a')
-    await new Promise(r => setTimeout(r, 100))
+    await delay(100)
     throw new Error('watty')
   }).catch(() => order.push('err'))
   return ['1aerr', order.join('')]

Original file line number	Diff line number	Diff line change
`@@ -83,6 +83,7 @@ function Backend({`
`83`	`83`	`backend.query.result.count = +x.toString('utf8', i + 1, x.length - 1)`
`84`	`84`	`if (x[i - 1] >= 65) {`
`85`	`85`	`backend.query.result.command = x.toString('utf8', 5, i)`
	`86`	`+ backend.query.result.state = state`
`86`	`87`	`break`
`87`	`88`	`}`
`88`	`89`	`}`