git ssb

1
const Port = require('./port.js')
2
const PARENT = Symbol('parent')
3
4
// decides which message to go firts
5
function messageArbiter (pairA, pairB) {
6
  const a = pairA[1].peek()
7
  const b = pairB[1].peek()
8
9
  if (!a) {
10
    return pairB
11
  } else if (!b) {
12
    return pairA
13
  }
14
15
  const aGasPrice = a.resources.gasPrice
16
  const bGasPrice = b.resources.gasPrice
17
  if (a.ticks !== b.ticks) {
18
    return a.ticks < b.ticks ? pairA : pairB
19
  } else if (aGasPrice === bGasPrice) {
20
    return a.hash() > b.hash() ? pairA : pairB
21
  } else {
22
    return aGasPrice > bGasPrice ? pairA : pairB
23
  }
24
}
25
26
module.exports = class PortManager {
27
  constructor (kernel) {
28
    this.kernel = kernel
29
    this.hypervisor = kernel._opts.hypervisor
30
    this.ports = kernel.state.ports
31
    this.parentPort = kernel._opts.parentPort
32
    this._portMap = new Map()
33
  }
34
35
  async start () {
36
    // map ports to thier id's
37
    let ports = Object.keys(this.ports).map(name => {
38
      const port = this.ports[name]
39
      this._mapPort(name, port)
40
    })
41
42
    // create the parent port
43
    await Promise.all(ports)
44
    const parentID = await this.hypervisor.generateID(this.kernel._opts.parentPort)
45
    this._portMap.set(parentID, new Port(PARENT))
46
  }
47
48
  async _mapPort (name, port) {
49
    const id = await this.hypervisor.generateID(port)
50
    port = new Port(name)
51
    this._portMap.set(id, port)
52
  }
53
54
  queue (message) {
55
    this._portMap.get(message.fromPort).queue(message)
56
  }
57
58
  set (name, port) {
59
    this.ports[name] = port
60
    return this._mapPort(name, port)
61
  }
62
63
  async get (port) {
64
    const id = await this.hypervisor.generateID(port)
65
    return this._portMap.get(id)
66
  }
67
68
  // waits till all ports have reached a threshold tick count
69
  async wait (threshold) {
70
    // find the ports that have a smaller tick count then the threshold tick count
71
    const unkownPorts = [...this._portMap].filter(([id, port]) => {
72
      return (port.hasSent || port.name === PARENT) && port.ticks < threshold
73
    })
74
75
    const promises = unkownPorts.map(async ([id, port]) => {
76
      const portObj = port.name === PARENT ? this.parentPort : this.ports[port.name]
77
      // update the port's tick count
78
      port.ticks = await this.hypervisor.wait(portObj, threshold)
79
    })
80
    return Promise.all(promises)
81
  }
82
83
  async getNextMessage (ticks) {
84
    await this.wait(ticks)
85
    const portMap = [...this._portMap].reduce(messageArbiter)
86
    if (portMap) {
87
      return portMap[1].shift()
88
    }
89
  }
90
}
91