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) {
7
    super()
8
    this._queue = []
9
    this.state = state
10
    this.Kernel = KernelContructor
11
    // set up the parent port
12
    const parentPort = new Port()
13
    parentPort.on('message', message => {
14
      this._recieveMessage(message)
15
    })
16
    // create the cache
17
    this.cache = new Map()
18
    this.cache.set(common.PARENT, parentPort)
19
  }
20
21
  _recieveMessage (message) {
22
    const index = this._queue.push(message) - 1
23
    this.emit('message', index)
24
  }
25
26
  async get (name) {
27
    let port = this.cache.get(name)
28
    if (!port) {
29
      port = new Port()
30
      port.on('message', message => {
31
        this._recieveMessage(message)
32
      })
33
      // create destination kernel
34
      const state = await this.state.get(name)
35
      const destKernel = new this.Kernel({
36
        state: state,
37
        parent: port
38
      })
39
40
      // shutdown the kernel when it is done doing it work
41
      destKernel.on('idle', () => {
42
        destKernel.shutdown()
43
        this.cache.delete(name)
44
      })
45
46
      // find and connect the destination port
47
      const destPort = await destKernel.ports.get(common.PARENT)
48
      port.connect(destPort)
49
      this.cache.set(name, port)
50
    }
51
    return port
52
  }
53
54
  async peek (index = 0) {
55
    return this._queue[index]
56
  }
57
58
  remove (index) {
59
    return this._queue.splice(index, index + 1)
60
  }
61
62
  close () {
63
    for (let port in this.cache) {
64
      port.emit('close')
65
    }
66
    this.cache.clear()
67
  }
68
}
69