Files: 818c7e2b325f1dac05933b54857443ddadc56c93 / actor.js
4184 bytesRaw
1 | const Pipe = require('buffer-pipe') |
2 | const Cap = require('primea-capability') |
3 | const Message = require('primea-message') |
4 | const leb128 = require('leb128').unsigned |
5 | const Inbox = require('./inbox.js') |
6 | |
7 | module.exports = class Actor { |
8 | /** |
9 | * the Actor manages the varous message passing functions and provides |
10 | * an interface for the containers to use |
11 | * @param {Object} opts |
12 | * @param {Object} opts.id - the UUID of the Actor |
13 | * @param {Object} opts.state - the state of the container |
14 | * @param {Object} opts.hypervisor - the instance of the hypervisor |
15 | * @param {Object} opts.container - the container constuctor and argments |
16 | */ |
17 | constructor (opts) { |
18 | Object.assign(this, opts) |
19 | |
20 | this.container = new opts.container.Constructor(this, opts.container.args) |
21 | this.inbox = new Inbox({ |
22 | actor: this, |
23 | hypervisor: opts.hypervisor |
24 | }) |
25 | |
26 | this.ticks = 0 |
27 | this.running = false |
28 | } |
29 | |
30 | /** |
31 | * Mints a new capabilitly with a given tag |
32 | * @param {*} tag - a tag which can be used to identify caps |
33 | * @return {Object} |
34 | */ |
35 | mintCap (tag = 0, funcIndex = 0) { |
36 | return new Cap(this.id, tag, funcIndex) |
37 | } |
38 | |
39 | /** |
40 | * adds a message to this actor's message queue |
41 | * @param {string} portName |
42 | * @param {object} message |
43 | */ |
44 | queue (message) { |
45 | this.inbox.queue(message) |
46 | |
47 | if (!this.running) { |
48 | this.running = true |
49 | this._startMessageLoop() |
50 | } |
51 | } |
52 | |
53 | /** |
54 | * runs the creation routine for the actor |
55 | * @param {object} message |
56 | * @returns {Promise} |
57 | */ |
58 | create (message) { |
59 | this.running = true |
60 | return this.runMessage(message, 'onCreation').then(() => { |
61 | this._startMessageLoop() |
62 | }) |
63 | } |
64 | |
65 | // waits for the next message |
66 | async _startMessageLoop () { |
67 | // this ensure we only every have one loop running at a time |
68 | while (1) { |
69 | const message = await this.inbox.nextMessage(0, true) |
70 | if (!message) break |
71 | |
72 | // run the next message |
73 | await this.runMessage(message) |
74 | } |
75 | this.running = false |
76 | // wait for state ops to finish |
77 | await this.state.done() |
78 | if (!this.running) { |
79 | this.container.onIdle() |
80 | } |
81 | } |
82 | |
83 | serializeMetaData () { |
84 | return Actor.serializeMetaData(this.type, this.nonce) |
85 | } |
86 | |
87 | static serializeMetaData (type, nonce = 0) { |
88 | const p = new Pipe() |
89 | leb128.write(type, p) |
90 | leb128.write(nonce, p) |
91 | return p.buffer |
92 | } |
93 | |
94 | static deserializeMetaData (buffer) { |
95 | const pipe = new Pipe(buffer) |
96 | const type = leb128.read(pipe) |
97 | const nonce = leb128.read(pipe) |
98 | return { |
99 | nonce, |
100 | type |
101 | } |
102 | } |
103 | |
104 | /** |
105 | * Runs the shutdown routine for the actor |
106 | */ |
107 | shutdown () { |
108 | this.state.root['/'][3] = this.serializeMetaData() |
109 | this.hypervisor.scheduler.done(this.id) |
110 | } |
111 | |
112 | /** |
113 | * Runs the startup routine for the actor |
114 | */ |
115 | startup () { |
116 | return this.container.onStartup() |
117 | } |
118 | |
119 | /** |
120 | * run the Actor with a given message |
121 | * @param {object} message - the message to run |
122 | * @param {String} method - which method to run |
123 | * @returns {Promise} |
124 | */ |
125 | async runMessage (message, method = 'onMessage') { |
126 | try { |
127 | await this.container[method](message) |
128 | } catch (e) { |
129 | message.emit('execution:error', e) |
130 | } |
131 | } |
132 | |
133 | /** |
134 | * updates the number of ticks that the actor has run |
135 | * @param {Number} count - the number of ticks to add |
136 | */ |
137 | incrementTicks (count) { |
138 | this.ticks += count |
139 | this.hypervisor.scheduler.update(this) |
140 | } |
141 | |
142 | /** |
143 | * creates an actor |
144 | * @param {Integer} type - the type id for the container |
145 | * @param {Object} message - an intial [message](https://github.com/primea/js-primea-message) to send newly created actor |
146 | */ |
147 | createActor (type, message) { |
148 | const id = this._generateNextId() |
149 | return this.hypervisor.createActor(type, message, id) |
150 | } |
151 | |
152 | _generateNextId () { |
153 | const id = { |
154 | nonce: this.nonce, |
155 | parent: this.id |
156 | } |
157 | |
158 | this.nonce++ |
159 | return id |
160 | } |
161 | |
162 | /** |
163 | * sends a message to a given port |
164 | * @param {Object} portRef - the port |
165 | * @param {Message} message - the message |
166 | */ |
167 | send (cap, message) { |
168 | message._fromTicks = this.ticks |
169 | message._fromId = this.id |
170 | |
171 | return this.hypervisor.send(cap, message) |
172 | } |
173 | } |
174 |
Built with git-ssb-web