git ssb

1
const Port = require('./port.js')
2
const BN = require('bn.js')
3
const ENTRY = Symbol('entry')
4
5
// decides which message to go first
6
function messageArbiter (pairA, pairB) {
7
  const portA = pairA[1]
8
  const portB = pairB[1]
9
  const a = portA.peek()
10
  const b = portB.peek()
11
12
  if (!a) {
13
    return pairB
14
  } else if (!b) {
15
    return pairA
16
  }
17
18
  // order by number of ticks if messages have different number of ticks
19
  if (portA.ticks !== portB.ticks) {
20
    return portA.ticks < portB.ticks ? pairA : pairB
21
  } else if (a.priority !== b.priority) {
22
    // decide by priority
23
    return a.priority > b.priority ? pairA : pairB
24
  } else {
25
    // insertion order
26
    return pairA
27
  }
28
}
29
30
module.exports = class PortManager {
31
  constructor (opts) {
32
    Object.assign(this, opts)
33
    this._portMap = new Map()
34
  }
35
36
  async start () {
37
    // skip the root, since it doesn't have a parent
38
    if (this.parentPort !== undefined) {
39
      this._bindRef(this.parentPort, ENTRY)
40
    }
41
42
    // map ports to thier id's
43
    this.ports = await this.hypervisor.graph.get(this.state, 'ports')
44
    Object.keys(this.ports).map(name => {
45
      const port = this.ports[name]
46
      this._bindRef(port, name)
47
    })
48
  }
49
50
  _bindRef (portRef, name) {
51
    const port = new Port(name)
52
    this._portMap.set(portRef, port)
53
  }
54
55
  bind (port, name) {
56
    // save the port instance
57
    this.ports[name] = port
58
    this._bindRef(port, name)
59
  }
60
61
  queue (message) {
62
    this._portMap.get(message.fromPort).queue(message)
63
  }
64
65
  get (key) {
66
    return this.ports[key]
67
  }
68
69
  delete (key) {
70
    const port = this.ports[key]
71
    delete this.ports[key]
72
    this._portMap.delete(port)
73
  }
74
75
  isValidPort (port) {
76
    return this._portMap.has(port)
77
  }
78
79
  create (type) {
80
    const Container = this.hypervisor._containerTypes[type]
81
    const parentId = this.entryPort ? this.entryPort.id : null
82
    let nonce = this.state['/'].nonce
83
84
    const portRef = {
85
      'messages': [],
86
      'id': {
87
        '/': {
88
          nonce: nonce,
89
          parent: parentId
90
        }
91
      },
92
      'type': type,
93
      'link': {
94
        '/': Container.createState()
95
      }
96
    }
97
98
    // incerment the nonce
99
    nonce = new BN(nonce)
100
    nonce.iaddn(1)
101
    this.state['/'].nonce = nonce.toArray()
102
    return portRef
103
  }
104
105
  // waits till all ports have reached a threshold tick count
106
  wait (threshold, fromPort = this.entryPort, ports = this._portMap) {
107
    // find the ports that have a smaller tick count then the threshold tick count
108
    const unkownPorts = [...ports].filter(([portRef, port]) => {
109
      return port.ticks < threshold && fromPort !== portRef
110
    })
111
112
    const promises = unkownPorts.map(async([portRef, port]) => {
113
      // update the port's tick count
114
      port.ticks = await this.hypervisor.wait(portRef, threshold, this.entryPort)
115
    })
116
117
    return Promise.all(promises)
118
  }
119
120
  async getNextMessage () {
121
    if (this._portMap.size) {
122
      // find the oldest message
123
      const ticks = [...this._portMap].map(([name, port]) => {
124
        return port.size ? port.ticks : this.exoInterface.ticks
125
      }).reduce((ticksA, ticksB) => {
126
        return ticksA < ticksB ? ticksA : ticksB
127
      })
128
129
      await this.wait(ticks)
130
      const portMap = [...this._portMap].reduce(messageArbiter)
131
      return portMap[1].shift()
132
    }
133
  }
134
}
135