| 78 | } |
| 79 | |
| 80 | async function init(sql, slot, publications) { |
| 81 | if (!publications) |
| 82 | throw new Error('Missing publication names') |
| 83 | |
| 84 | const xs = await sql.unsafe( |
| 85 | `CREATE_REPLICATION_SLOT ${ slot } TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT` |
| 86 | ) |
| 87 | |
| 88 | const [x] = xs |
| 89 | |
| 90 | const stream = await sql.unsafe( |
| 91 | `START_REPLICATION SLOT ${ slot } LOGICAL ${ |
| 92 | x.consistent_point |
| 93 | } (proto_version '1', publication_names '${ publications }')` |
| 94 | ).writable() |
| 95 | |
| 96 | const state = { |
| 97 | lsn: Buffer.concat(x.consistent_point.split('/').map(x => Buffer.from(('00000000' + x).slice(-8), 'hex'))) |
| 98 | } |
| 99 | |
| 100 | stream.on('data', data) |
| 101 | stream.on('error', error) |
| 102 | stream.on('close', sql.close) |
| 103 | |
| 104 | return { stream, state: xs.state } |
| 105 | |
| 106 | function error(e) { |
| 107 | console.error('Unexpected error during logical streaming - reconnecting', e) // eslint-disable-line |
| 108 | } |
| 109 | |
| 110 | function data(x) { |
| 111 | if (x[0] === 0x77) { |
| 112 | parse(x.subarray(25), state, sql.options.parsers, handle, options.transform) |
| 113 | } else if (x[0] === 0x6b && x[17]) { |
| 114 | state.lsn = x.subarray(1, 9) |
| 115 | pong() |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | function handle(a, b) { |
| 120 | const path = b.relation.schema + '.' + b.relation.table |
| 121 | call('*', a, b) |
| 122 | call('*:' + path, a, b) |
| 123 | b.relation.keys.length && call('*:' + path + '=' + b.relation.keys.map(x => a[x.name]), a, b) |
| 124 | call(b.command, a, b) |
| 125 | call(b.command + ':' + path, a, b) |
| 126 | b.relation.keys.length && call(b.command + ':' + path + '=' + b.relation.keys.map(x => a[x.name]), a, b) |
| 127 | } |
| 128 | |
| 129 | function pong() { |
| 130 | const x = Buffer.alloc(34) |
| 131 | x[0] = 'r'.charCodeAt(0) |
| 132 | x.fill(state.lsn, 1) |
| 133 | x.writeBigInt64BE(BigInt(Date.now() - Date.UTC(2000, 0, 1)) * BigInt(1000), 25) |
| 134 | stream.write(x) |
| 135 | } |
| 136 | } |
| 137 | |