git ssb

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