git ssb

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