build

cjs/src/index.js

@@ -83,7 +83,7 @@ function Postgres(a, b) {
 
   return sql
 
-  function Sql(handler, instant) {
+  function Sql(handler) {
     handler.debug = options.debug
 
     Object.entries(options.types).reduce((acc, [name, type]) => {
@@ -112,7 +112,6 @@ function Postgres(a, b) {
         : typeof strings === 'string' && !args.length
           ? new Identifier(options.transform.column.to ? options.transform.column.to(strings) : strings)
           : new Builder(strings, args)
-      instant && query instanceof Query && query.execute()
       return query
     }
 
@@ -123,7 +122,6 @@ function Postgres(a, b) {
         ...options,
         simple: 'simple' in options ? options.simple : args.length === 0
       })
-      instant && query.execute()
       return query
     }
 
@@ -141,7 +139,6 @@ function Postgres(a, b) {
         ...options,
         simple: 'simple' in options ? options.simple : args.length === 0
       })
-      instant && query.execute()
       return query
     }
   }

cjs/src/subscribe.js

@@ -97,11 +97,15 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
     }
 
     stream.on('data', data)
-    stream.on('error', sql.close)
+    stream.on('error', error)
     stream.on('close', sql.close)
 
     return { stream, state: xs.state }
 
+    function error(e) {
+      console.error('Unexpected error during logical streaming - reconnecting', e)
+    }
+
     function data(x) {
       if (x[0] === 0x77)
         parse(x.subarray(25), state, sql.options.parsers, handle, options.transform)
@@ -191,7 +195,7 @@ function parse(x, state, parsers, handle, transform) {
       i += 4
       const key = x[i] === 75
       handle(key || x[i] === 79
-        ? tuples(x, key ? relation.keys : relation.columns, i += 3, transform).row
+        ? tuples(x, relation.columns, i += 3, transform).row
         : null
       , {
         command: 'delete',
@@ -205,7 +209,7 @@ function parse(x, state, parsers, handle, transform) {
       i += 4
       const key = x[i] === 75
       const xs = key || x[i] === 79
-        ? tuples(x, key ? relation.keys : relation.columns, i += 3, transform)
+        ? tuples(x, relation.columns, i += 3, transform)
         : null
 
       xs && (i = xs.i)

cjs/tests/index.js

@@ -1989,9 +1989,9 @@ t('subscribe', { timeout: 2 }, async() => {
 
   const result = []
 
-  const { unsubscribe } = await sql.subscribe('*', (row, { command, old }) =>
-    result.push(command, row.name || row.id, old && old.name)
-  )
+  const { unsubscribe } = await sql.subscribe('*', (row, { command, old }) => {
+    result.push(command, row.name, row.id, old && old.name, old && old.id)
+  })
 
   await sql`
     create table test (
@@ -2003,6 +2003,7 @@ t('subscribe', { timeout: 2 }, async() => {
   await sql`alter table test replica identity default`
   await sql`insert into test (name) values ('Murray')`
   await sql`update test set name = 'Rothbard'`
+  await sql`update test set id = 2`
   await sql`delete from test`
   await sql`alter table test replica identity full`
   await sql`insert into test (name) values ('Murray')`
@@ -2013,7 +2014,7 @@ t('subscribe', { timeout: 2 }, async() => {
   await sql`insert into test (name) values ('Oh noes')`
   await delay(10)
   return [
-    'insert,Murray,,update,Rothbard,,delete,1,,insert,Murray,,update,Rothbard,Murray,delete,Rothbard,',
+    'insert,Murray,1,,,update,Rothbard,1,,,update,Rothbard,2,,1,delete,,2,,,insert,Murray,2,,,update,Rothbard,2,Murray,2,delete,Rothbard,2,,',
     result.join(','),
     await sql`drop table test`,
     await sql`drop publication alltables`,
@@ -2139,7 +2140,7 @@ t('Cancel queued query', async() => {
   const query = sql`select pg_sleep(2) as nej`
   const tx = sql.begin(sql => (
     query.cancel(),
-    sql`select pg_sleep(0.1) as hej, 'hejsa'`
+    sql`select pg_sleep(0.5) as hej, 'hejsa'`
   ))
   const error = await query.catch(x => x)
   await tx

deno/src/index.js

@@ -84,7 +84,7 @@ function Postgres(a, b) {
 
   return sql
 
-  function Sql(handler, instant) {
+  function Sql(handler) {
     handler.debug = options.debug
 
     Object.entries(options.types).reduce((acc, [name, type]) => {
@@ -113,7 +113,6 @@ function Postgres(a, b) {
         : typeof strings === 'string' && !args.length
           ? new Identifier(options.transform.column.to ? options.transform.column.to(strings) : strings)
           : new Builder(strings, args)
-      instant && query instanceof Query && query.execute()
       return query
     }
 
@@ -124,7 +123,6 @@ function Postgres(a, b) {
         ...options,
         simple: 'simple' in options ? options.simple : args.length === 0
       })
-      instant && query.execute()
       return query
     }
 
@@ -142,7 +140,6 @@ function Postgres(a, b) {
         ...options,
         simple: 'simple' in options ? options.simple : args.length === 0
       })
-      instant && query.execute()
       return query
     }
   }

deno/src/subscribe.js

@@ -98,11 +98,15 @@ export default function Subscribe(postgres, options) {
     }
 
     stream.on('data', data)
-    stream.on('error', sql.close)
+    stream.on('error', error)
     stream.on('close', sql.close)
 
     return { stream, state: xs.state }
 
+    function error(e) {
+      console.error('Unexpected error during logical streaming - reconnecting', e)
+    }
+
     function data(x) {
       if (x[0] === 0x77)
         parse(x.subarray(25), state, sql.options.parsers, handle, options.transform)
@@ -192,7 +196,7 @@ function parse(x, state, parsers, handle, transform) {
       i += 4
       const key = x[i] === 75
       handle(key || x[i] === 79
-        ? tuples(x, key ? relation.keys : relation.columns, i += 3, transform).row
+        ? tuples(x, relation.columns, i += 3, transform).row
         : null
       , {
         command: 'delete',
@@ -206,7 +210,7 @@ function parse(x, state, parsers, handle, transform) {
       i += 4
       const key = x[i] === 75
       const xs = key || x[i] === 79
-        ? tuples(x, key ? relation.keys : relation.columns, i += 3, transform)
+        ? tuples(x, relation.columns, i += 3, transform)
         : null
 
       xs && (i = xs.i)

deno/tests/index.js

@@ -1991,9 +1991,9 @@ t('subscribe', { timeout: 2 }, async() => {
 
   const result = []
 
-  const { unsubscribe } = await sql.subscribe('*', (row, { command, old }) =>
-    result.push(command, row.name || row.id, old && old.name)
-  )
+  const { unsubscribe } = await sql.subscribe('*', (row, { command, old }) => {
+    result.push(command, row.name, row.id, old && old.name, old && old.id)
+  })
 
   await sql`
     create table test (
@@ -2005,6 +2005,7 @@ t('subscribe', { timeout: 2 }, async() => {
   await sql`alter table test replica identity default`
   await sql`insert into test (name) values ('Murray')`
   await sql`update test set name = 'Rothbard'`
+  await sql`update test set id = 2`
   await sql`delete from test`
   await sql`alter table test replica identity full`
   await sql`insert into test (name) values ('Murray')`
@@ -2015,7 +2016,7 @@ t('subscribe', { timeout: 2 }, async() => {
   await sql`insert into test (name) values ('Oh noes')`
   await delay(10)
   return [
-    'insert,Murray,,update,Rothbard,,delete,1,,insert,Murray,,update,Rothbard,Murray,delete,Rothbard,',
+    'insert,Murray,1,,,update,Rothbard,1,,,update,Rothbard,2,,1,delete,,2,,,insert,Murray,2,,,update,Rothbard,2,Murray,2,delete,Rothbard,2,,',
     result.join(','),
     await sql`drop table test`,
     await sql`drop publication alltables`,
@@ -2141,7 +2142,7 @@ t('Cancel queued query', async() => {
   const query = sql`select pg_sleep(2) as nej`
   const tx = sql.begin(sql => (
     query.cancel(),
-    sql`select pg_sleep(0.1) as hej, 'hejsa'`
+    sql`select pg_sleep(0.5) as hej, 'hejsa'`
   ))
   const error = await query.catch(x => x)
   await tx