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