Files: cffaab0554c3a8228f13ad2e78df2cc6aab77296 / actor.js
2893 bytesRaw
1 | const Pipe = require('buffer-pipe') |
2 | const leb128 = require('leb128').unsigned |
3 | |
4 | module.exports = class Actor { |
5 | /** |
6 | * the Actor manages the varous message passing functions and provides |
7 | * an interface for the containers to use |
8 | * @param {Object} opts |
9 | * @param {Object} opts.id - the UUID of the Actor |
10 | * @param {Object} opts.state - the state of the container |
11 | * @param {Object} opts.hypervisor - the instance of the hypervisor |
12 | * @param {Object} opts.container - the container constuctor and argments |
13 | */ |
14 | constructor (opts) { |
15 | Object.assign(this, opts) |
16 | |
17 | this.inbox = [] |
18 | this.ticks = 0 |
19 | this.running = false |
20 | this.container = new this.Container(this) |
21 | } |
22 | |
23 | serializeMetaData () { |
24 | return Actor.serializeMetaData(this.type, this.nonce) |
25 | } |
26 | |
27 | getFuncRef (name) { |
28 | return { |
29 | name, |
30 | destId: this.id |
31 | } |
32 | } |
33 | |
34 | /** |
35 | * Runs the shutdown routine for the actor |
36 | */ |
37 | async shutdown () { |
38 | await this.state.done() |
39 | this.state.root['/'][3] = this.serializeMetaData() |
40 | } |
41 | |
42 | /** |
43 | * Runs the startup routine for the actor |
44 | */ |
45 | startup () { |
46 | return this.container.onStartup() |
47 | } |
48 | |
49 | /** |
50 | * run the Actor with a given message |
51 | * @param {object} message - the message to run |
52 | * @param {String} method - which method to run |
53 | * @returns {Promise} |
54 | */ |
55 | async runMessage (message) { |
56 | if (message._fromTicks > this.ticks) { |
57 | this.ticks = message._fromTicks |
58 | } |
59 | try { |
60 | this.currentMessage = message |
61 | await this.container.onMessage(message) |
62 | } catch (e) { |
63 | message.emit('execution:error', e) |
64 | } |
65 | message.emit('done') |
66 | } |
67 | |
68 | /** |
69 | * updates the number of ticks that the actor has run |
70 | * @param {Number} count - the number of ticks to add |
71 | */ |
72 | incrementTicks (count) { |
73 | this.ticks += count |
74 | } |
75 | |
76 | /** |
77 | * creates an actor |
78 | * @param {Integer} type - the type id for the container |
79 | * @param {Object} message - an intial [message](https://github.com/primea/js-primea-message) to send newly created actor |
80 | */ |
81 | createActor (type, code) { |
82 | const id = this._generateNextId() |
83 | return this.hypervisor.createActor(type, code, id) |
84 | } |
85 | |
86 | _generateNextId () { |
87 | const id = { |
88 | nonce: this.nonce, |
89 | parent: this.id |
90 | } |
91 | |
92 | this.nonce++ |
93 | return id |
94 | } |
95 | |
96 | /** |
97 | * sends a message to a given port |
98 | * @param {Object} portRef - the port |
99 | * @param {Message} message - the message |
100 | */ |
101 | send (message) { |
102 | message._fromTicks = this.ticks |
103 | message._fromId = this.id |
104 | |
105 | this.hypervisor.scheduler.queue([message]) |
106 | } |
107 | |
108 | static serializeMetaData (type, nonce = 0) { |
109 | const p = new Pipe() |
110 | leb128.write(type, p) |
111 | leb128.write(nonce, p) |
112 | return p.buffer |
113 | } |
114 | |
115 | static deserializeMetaData (buffer) { |
116 | const pipe = new Pipe(buffer) |
117 | const type = leb128.read(pipe) |
118 | const nonce = leb128.read(pipe) |
119 | return { |
120 | nonce, |
121 | type |
122 | } |
123 | } |
124 | } |
125 |
Built with git-ssb-web