Files: c0ffdd0e09232954459f879b99db0b4192eded09 / portManager.js
1918 bytesRaw
1 | const EventEmitter = require('events') |
2 | const path = require('path') |
3 | const Port = require('./port.js') |
4 | const common = require('./common.js') |
5 | |
6 | module.exports = class PortManager extends EventEmitter { |
7 | constructor (opts) { |
8 | super() |
9 | Object.assign(this, opts) |
10 | this._queue = [] |
11 | // set up the parent port |
12 | const parentPort = new Port(common.PARENT) |
13 | parentPort.on('message', message => { |
14 | this._recieveMessage(message) |
15 | }) |
16 | |
17 | // create the cache |
18 | this.cache = new Map() |
19 | this.cache.set(common.PARENT, parentPort) |
20 | } |
21 | |
22 | _recieveMessage (message) { |
23 | const index = this._queue.push(message) - 1 |
24 | this.emit('message', index) |
25 | } |
26 | |
27 | async get (name) { |
28 | let port = this.cache.get(name) |
29 | if (!port) { |
30 | port = new Port(name) |
31 | port.on('message', message => { |
32 | this._recieveMessage(message) |
33 | }) |
34 | // create destination kernel |
35 | const state = await this.graph.get(this.state, name) |
36 | |
37 | const destKernel = new this.Kernel({ |
38 | state: state, |
39 | graph: this.graph, |
40 | parentPort: port, |
41 | imports: this.imports, |
42 | path: path.join(this.path, name) |
43 | }) |
44 | |
45 | // shutdown the kernel when it is done doing it work |
46 | destKernel.on('idle', () => { |
47 | destKernel.shutdown() |
48 | this.cache.delete(name) |
49 | }) |
50 | |
51 | // find and connect the destination port |
52 | const destPort = await destKernel.ports.get(common.PARENT) |
53 | port.connect(destPort) |
54 | this.cache.set(name, port) |
55 | } |
56 | return port |
57 | } |
58 | |
59 | async peek (index = 0) { |
60 | return this._queue[index] |
61 | } |
62 | |
63 | remove (index) { |
64 | return this._queue.splice(index, index + 1) |
65 | } |
66 | |
67 | async send (message) { |
68 | let portName = message.nextPort() |
69 | const port = await this.get(portName) |
70 | port.send(message) |
71 | return message.result() |
72 | } |
73 | |
74 | close () { |
75 | for (let port in this.cache) { |
76 | port.emit('close') |
77 | } |
78 | this.cache.clear() |
79 | } |
80 | } |
81 |
Built with git-ssb-web