Files: 6f92591fea03c67a9e18c1daab07a4ed7d693b21 / actor.js
2941 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 | // console.log(this.container.onMessage) |
62 | await this.container.onMessage(message) |
63 | } catch (e) { |
64 | message.emit('execution:error', e) |
65 | } |
66 | message.emit('done') |
67 | } |
68 | |
69 | /** |
70 | * updates the number of ticks that the actor has run |
71 | * @param {Number} count - the number of ticks to add |
72 | */ |
73 | incrementTicks (count) { |
74 | this.ticks += count |
75 | } |
76 | |
77 | /** |
78 | * creates an actor |
79 | * @param {Integer} type - the type id for the container |
80 | * @param {Object} message - an intial [message](https://github.com/primea/js-primea-message) to send newly created actor |
81 | */ |
82 | createActor (type, code) { |
83 | const id = this._generateNextId() |
84 | return this.hypervisor.createActor(type, code, id) |
85 | } |
86 | |
87 | _generateNextId () { |
88 | const id = { |
89 | nonce: this.nonce, |
90 | parent: this.id |
91 | } |
92 | |
93 | this.nonce++ |
94 | return id |
95 | } |
96 | |
97 | /** |
98 | * sends a message to a given port |
99 | * @param {Object} portRef - the port |
100 | * @param {Message} message - the message |
101 | */ |
102 | send (message) { |
103 | message._fromTicks = this.ticks |
104 | message._fromId = this.id |
105 | |
106 | this.hypervisor.scheduler.queue([message]) |
107 | } |
108 | |
109 | static serializeMetaData (type, nonce = 0) { |
110 | const p = new Pipe() |
111 | leb128.write(type, p) |
112 | leb128.write(nonce, p) |
113 | return p.buffer |
114 | } |
115 | |
116 | static deserializeMetaData (buffer) { |
117 | const pipe = new Pipe(buffer) |
118 | const type = leb128.read(pipe) |
119 | const nonce = leb128.read(pipe) |
120 | return { |
121 | nonce, |
122 | type |
123 | } |
124 | } |
125 | } |
126 |
Built with git-ssb-web