git ssb

1
const EventEmitter = require('events')
2
const Vertex = require('merkle-trie')
3
const PortManager = require('./portManager.js')
4
const StateInterface = require('./stateInterface.js')
5
const imports = require('./EVMinterface.js')
6
const codeHandler = require('./codeHandler.js')
7
const common = require('./common.js')
8
9
module.exports = class Kernel extends EventEmitter {
10
  constructor (opts = {}) {
11
    super()
12
    const state = this.state = opts.state || new Vertex()
13
    this.stateInterface = new StateInterface(state)
14
    this.code = opts.code || state.value
15
    this.path = state.path
16
    this.imports = opts.imports || [imports]
17
    this.ports = new PortManager(state, opts.parent, Kernel)
18
    this._sentAtomicMessages = []
19
    this._vm = (opts.codeHandler || codeHandler).init(this.code)
20
    this._state = 'idle'
21
    this.ports.on('message', index => {
22
      this.runNextMessage(index)
23
    })
24
  }
25
26
  runNextMessage (index = 0) {
27
    return this.ports.peek(index).then(message => {
28
      if (message && (message.isCyclic(this) || this._state === 'idle')) {
29
        this.ports.remove(index)
30
        return this.run(message)
31
      } else {
32
        this._state = 'idle'
33
        this.emit('idle')
34
      }
35
    })
36
  }
37
38
  /**
39
   * run the kernels code with a given enviroment
40
   * The Kernel Stores all of its state in the Environment. The Interface is used
41
   * to by the VM to retrive infromation from the Environment.
42
   */
43
  async run (message, imports = this.imports) {
44
    function revert () {
45
      // revert the state
46
      this.state.set([], oldState)
47
      // revert all the sent messages
48
      for (let msg in this._sentAtomicMessages) {
49
        msg.revert()
50
      }
51
      this.runNextMessage(0)
52
    }
53
54
    const oldState = this.state.copy()
55
    let result
56
    this._state = 'running'
57
    try {
58
      result = await this._vm.run(message, this, imports) || {}
59
    } catch (e) {
60
      console.log(e)
61
      result = {
62
        exception: true
63
      }
64
    }
65
    if (result.execption) {
66
      // failed messages
67
      revert()
68
    } else if (message.atomic) {
69
      // messages
70
      message._finish()
71
      message.result().then(result => {
72
        if (result.execption) {
73
          revert()
74
        } else {
75
          this.runNextMessage(0)
76
        }
77
      })
78
79
      if (message.hops === message.to.length) {
80
        message._respond(result)
81
      }
82
    } else {
83
      // non-atomic messages
84
      this.runNextMessage(0)
85
    }
86
    return result
87
  }
88
89
  async send (message) {
90
    let portName = message.nextPort()
91
    // replace root with parent path to root
92
    if (portName === common.ROOT) {
93
      message.to.shift()
94
      message.to = new Array(this.path.length).fill(common.PARENT).concat(message.to)
95
      portName = common.PARENT
96
    }
97
    message.addVistedKernel(this)
98
    this.lastMessage = message
99
    // console.log(portName, message)
100
    const port = await this.ports.get(portName)
101
    // save the atomic messages for possible reverts
102
    if (message.atomic) {
103
      this._sentAtomicMessages.push(message)
104
    }
105
    port.send(message)
106
    return message.result()
107
  }
108
109
  shutdown () {}
110
}
111