| 1 | import { Readable } from 'stream' |
| 2 | import _ from 'lodash' |
| 3 | import { LIST, wantArray } from './cross' |
| 4 | import { Context } from 'koa' |
| 5 | import events, { OnOptions } from './events' |
| 6 | |
| 7 | type SendListFunc<T> = (list:SendListReadable<T>) => void |
| 8 | // offer an api for a generic dynamic list. Suitable to be the result of an api. |
| 9 | export class SendListReadable<T> extends Readable { |
| 10 | protected lastError: string | number | undefined |
| 11 | protected buffer: any[] = [] |
| 12 | protected processBuffer: _.DebouncedFunc<any> |
| 13 | protected sent: undefined | T[] |
| 14 | constructor({ addAtStart, doAtStart, bufferTime, onEnd, diff }: |
| 15 | { bufferTime?: number, addAtStart?: T[], doAtStart?: SendListFunc<T>, onEnd?: SendListFunc<T>, diff?: boolean }={}) { |
| 16 | super({ objectMode: true, read(){} }) |
| 17 | if (!bufferTime) |
| 18 | bufferTime = 200 |
| 19 | if (diff) |
| 20 | this.sent = [] |
| 21 | this.processBuffer = _.debounce(() => { |
| 22 | const {sent} = this |
| 23 | if (sent) |
| 24 | this.buffer = this.buffer.filter(([cmd, a, b]) => { |
| 25 | if (cmd === LIST.add) |
| 26 | return sent.push(...wantArray(a)) |
| 27 | if (cmd === LIST.remove) |
| 28 | return _.remove(sent, a) |
| 29 | if (cmd !== LIST.update) |
| 30 | return true |
| 31 | const found = _.find(sent, a) as any |
| 32 | if (!found) return |
| 33 | for (const k in b) |
| 34 | if (b[k] === found[k]) |
| 35 | delete b[k] |
| 36 | else { |
| 37 | found[k] = b[k] |
| 38 | b[k] ??= null // go and delete it, remotely |
| 39 | } |
| 40 | return !_.isEmpty(b) |
| 41 | }) |
| 42 | if (!this.buffer.length) return |
| 43 | this.push(this.buffer) |
| 44 | this.buffer = [] |
| 45 | }, bufferTime, { maxWait: bufferTime }) |
| 46 | this.on('close', () => { |
| 47 | onEnd?.(this) |
| 48 | this.destroy() |
| 49 | }) |
| 50 | setTimeout(() => doAtStart?.(this)) // work later, when list object has been received by Koa |
| 51 | if (addAtStart) { |
| 52 | for (const x of addAtStart) |
| 53 | this.add(x) |
| 54 | this.ready() |
| 55 | } |
| 56 | } |
| 57 | protected _push(rec: any) { |
| 58 | this.buffer.push(rec) |
| 59 | if (this.buffer.length > 10_000) // hard limit |
| 60 | this.processBuffer.flush() |
| 61 | else |
| 62 | this.processBuffer() |
| 63 | } |
| 64 | add(rec: T) { |
| 65 | this._push([LIST.add, rec]) |
| 66 | } |
| 67 | remove(search: Partial<T>) { |
| 68 | const match = _.matches(search) |
| 69 | const idx = _.findIndex(this.buffer, x => match(x[1])) |
| 70 | const found = this.buffer[idx] |
| 71 | const op = found?.[0] |
| 72 | if (op === LIST.remove) return |
| 73 | if (found) { |
| 74 | this.buffer.splice(idx, 1) |
| 75 | if (op === LIST.add) return // assuming this never reached the client |
| 76 | } |
| 77 | this._push([LIST.remove, search]) |
| 78 | } |
| 79 | update(search: Partial<T>, change: Partial<T>) { |
| 80 | if (_.isEmpty(change)) return |
| 81 | const match = _.matches(search) |
| 82 | const found = _.find(this.buffer, x => match(x[1])) |
| 83 | const op = found?.[0] |
| 84 | if (op === LIST.remove) return |
| 85 | if (op === LIST.add || op === LIST.update) |
| 86 | return Object.assign(found[op === LIST.add ? 1 : 2], change) |
| 87 | this._push([LIST.update, search, change]) |
| 88 | } |
| 89 | ready() { // useful to indicate the end of an initial phase, but we leave open for updates |
| 90 | this._push([LIST.ready]) |
| 91 | } |
| 92 | custom(name: string, data: any) { |
| 93 | this._push(data === undefined ? [name] : [name, data]) |
| 94 | } |
| 95 | props(props: object) { |
| 96 | this._push([LIST.props, props]) |
| 97 | } |
| 98 | error(msg: NonNullable<typeof this.lastError>, close=false, props?: object) { |
| 99 | this._push([LIST.error, msg, props]) |
| 100 | this.lastError = msg |
| 101 | if (close) |
| 102 | this.close() |
| 103 | } |
| 104 | getLastError() { |
| 105 | return this.lastError |
| 106 | } |
| 107 | close() { |
| 108 | this.processBuffer.flush() |
| 109 | this.push(null) |
| 110 | } |
| 111 | events(ctx: Context, eventMap: Parameters<typeof events.multi>[0], options?: OnOptions) { |
| 112 | ctx.res.once('close', events.multi(eventMap, options)) |
| 113 | return this |
| 114 | } |
| 115 | isClosed() { |
| 116 | return this.destroyed |
| 117 | } |
| 118 | } |