git ssb

1
const Port = require('./port.js')
2
const ENTRY = Symbol('entry')
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 (opts) {
28
    Object.assign(this, opts)
29
    this._portMap = new Map()
30
  }
31
32
  async start () {
33
    // map ports to thier id's
34
    this.ports = await this.hypervisor.graph.get(this.state, 'ports')
35
    Object.keys(this.ports).map(name => {
36
      const port = this.ports[name]
37
      this._mapPort(name, port)
38
    })
39
40
    // skip the root, since it doesn't have a parent
41
    if (this.parentPort !== undefined) {
42
      this._portMap.set(this.parentPort, new Port(ENTRY))
43
    }
44
  }
45
46
  _mapPort (name, portRef) {
47
    const port = new Port(name)
48
    this._portMap.set(portRef, port)
49
  }
50
51
  queue (message) {
52
    this._portMap.get(message.fromPort).queue(message)
53
  }
54
55
  set (name, port) {
56
    this.ports[name] = port
57
    return this._mapPort(name, port)
58
  }
59
60
  get (port) {
61
    return this._portMap.get(port)
62
  }
63
64
  getRef (key) {
65
    if (key === ENTRY) {
66
      return this.entryPort
67
    } else {
68
      return this.ports[key]
69
    }
70
  }
71
72
  // waits till all ports have reached a threshold tick count
73
  wait (threshold, fromPort) {
74
    // find the ports that have a smaller tick count then the threshold tick count
75
    const unkownPorts = [...this._portMap].filter(([portRef, port]) => {
76
      return port.ticks < threshold && fromPort !== portRef
77
    })
78
79
    const promises = unkownPorts.map(async ([portRef, port]) => {
80
      // update the port's tick count
81
      port.ticks = await this.hypervisor.wait(portRef, threshold, this.entryPort)
82
    })
83
    return Promise.all(promises)
84
  }
85
86
  async getNextMessage () {
87
    await this.wait(this.kernel.ticks, this.entryPort)
88
    const portMap = [...this._portMap].reduce(messageArbiter)
89
    if (portMap) {
90
      return portMap[1].shift()
91
    }
92
  }
93
}
94