| 1 | const Stream = require('stream') |
| 2 | |
| 3 | module.exports = largeObject;function largeObject(sql, oid, mode = 0x00020000 | 0x00040000) { |
| 4 | return new Promise(async(resolve, reject) => { |
| 5 | await sql.begin(async sql => { |
| 6 | let finish |
| 7 | !oid && ([{ oid }] = await sql`select lo_creat(-1) as oid`) |
| 8 | const [{ fd }] = await sql`select lo_open(${ oid }, ${ mode }) as fd` |
| 9 | |
| 10 | const lo = { |
| 11 | writable, |
| 12 | readable, |
| 13 | close : () => sql`select lo_close(${ fd })`.then(finish), |
| 14 | tell : () => sql`select lo_tell64(${ fd })`, |
| 15 | read : (x) => sql`select loread(${ fd }, ${ x }) as data`, |
| 16 | write : (x) => sql`select lowrite(${ fd }, ${ x })`, |
| 17 | truncate : (x) => sql`select lo_truncate64(${ fd }, ${ x })`, |
| 18 | seek : (x, whence = 0) => sql`select lo_lseek64(${ fd }, ${ x }, ${ whence })`, |
| 19 | size : () => sql` |
| 20 | select |
| 21 | lo_lseek64(${ fd }, location, 0) as position, |
| 22 | seek.size |
| 23 | from ( |
| 24 | select |
| 25 | lo_lseek64($1, 0, 2) as size, |
| 26 | tell.location |
| 27 | from (select lo_tell64($1) as location) tell |
| 28 | ) seek |
| 29 | ` |
| 30 | } |
| 31 | |
| 32 | resolve(lo) |
| 33 | |
| 34 | return new Promise(async r => finish = r) |
| 35 | |
| 36 | async function readable({ |
| 37 | highWaterMark = 2048 * 8, |
| 38 | start = 0, |
| 39 | end = Infinity |
| 40 | } = {}) { |
| 41 | let max = end - start |
| 42 | start && await lo.seek(start) |
| 43 | return new Stream.Readable({ |
| 44 | highWaterMark, |
| 45 | async read(size) { |
| 46 | const l = size > max ? size - max : size |
| 47 | max -= size |
| 48 | const [{ data }] = await lo.read(l) |
| 49 | this.push(data) |
| 50 | if (data.length < size) |
| 51 | this.push(null) |
| 52 | } |
| 53 | }) |
| 54 | } |
| 55 | |
| 56 | async function writable({ |
| 57 | highWaterMark = 2048 * 8, |
| 58 | start = 0 |
| 59 | } = {}) { |
| 60 | start && await lo.seek(start) |