| 19 | } |
| 20 | |
| 21 | function createIpc<TIncoming, TOutgoing>( |
| 22 | port: number |
| 23 | ): Ipc<TIncoming, TOutgoing> { |
| 24 | const socket = createConnection({ |
| 25 | port, |
| 26 | host: '127.0.0.1', |
| 27 | }) |
| 28 | |
| 29 | /** |
| 30 | * A writable stream that writes to the socket. |
| 31 | * We don't write directly to the socket because we need to |
| 32 | * handle backpressure and wait for the socket to be drained |
| 33 | * before writing more data. |
| 34 | */ |
| 35 | const socketWritable = new Writable({ |
| 36 | write(chunk, _enc, cb) { |
| 37 | if (socket.write(chunk)) { |
| 38 | cb() |
| 39 | } else { |
| 40 | socket.once('drain', cb) |
| 41 | } |
| 42 | }, |
| 43 | final(cb) { |
| 44 | socket.end(cb) |
| 45 | }, |
| 46 | }) |
| 47 | |
| 48 | const packetQueue: Buffer[] = [] |
| 49 | const recvPromiseResolveQueue: Array<(message: TIncoming) => void> = [] |
| 50 | |
| 51 | function pushPacket(packet: Buffer) { |
| 52 | const recvPromiseResolve = recvPromiseResolveQueue.shift() |
| 53 | if (recvPromiseResolve != null) { |
| 54 | recvPromiseResolve(JSON.parse(packet.toString('utf8')) as TIncoming) |
| 55 | } else { |
| 56 | packetQueue.push(packet) |
| 57 | } |
| 58 | } |
| 59 | |
| 60 | let state: State = { type: 'waiting' } |
| 61 | let buffer: Buffer = Buffer.alloc(0) |
| 62 | socket.once('connect', () => { |
| 63 | socket.setNoDelay(true) |
| 64 | socket.on('data', (chunk) => { |
| 65 | buffer = Buffer.concat([buffer, chunk]) |
| 66 | |
| 67 | loop: while (true) { |
| 68 | switch (state.type) { |
| 69 | case 'waiting': { |
| 70 | if (buffer.length >= 4) { |
| 71 | const length = buffer.readUInt32BE(0) |
| 72 | buffer = buffer.subarray(4) |
| 73 | state = { type: 'packet', length } |
| 74 | } else { |
| 75 | break loop |
| 76 | } |
| 77 | break |
| 78 | } |