Actor system

A basic actor system only using a queue and a lock.

Usage

$3

``typescript
import * as Actor from "@yingyeothon/actor-system";
import * as InMemorySupport from "@yingyeothon/actor-system/lib/support/inmemory";

const subsys = {
queue: new InMemorySupport.InMemoryQueue(),
lock: new InMemorySupport.InMemoryLock(),
awaiter: new InMemorySupport.InMemoryAwaiter(),
};

class Adder {
private value = 0;

constructor(public readonly id: string) {}

public onMessage = (message: { delta: number }) => {
this.value += message.delta;
};
}

const env = { ...Actor.singleConsumer, ...subsys, ...new Adder(adder-1) };

// send means that produces a message to an actor and try to process it if it is possible.
// If other thread attaches this actor, it would process my message, too.
await Actor.send(env, { item: { delta: 1 } });
await Actor.send(env, { item: { delta: 2 } });
await Actor.send(env, { item: { delta: 3 } });
await Actor.send(env, { item: { delta: 4 } });

// adder actor would process all messages sequentially in background.
`

$3

It supports awaitPolicy that determines how long I should wait. Forget is default that means I don't want to wait anymore. We can choose Act that waits after onMesssage call or Commit that waits after onCommit call. And in that cases, awaitTimeoutMillis makes a timeout to wait.

`typescript
Actor.send(env, {
item: { delta: 10 },
awaitPolicy: Actor.AwaitPolicy.Act,
awaitTimeoutMillis: 100,
})
.then(/ HAPPY /) // It would be called after onMessage.
.catch(/ SAD /);
`

$3

It is too hard that an actor loads its context everytime to process onMessage. If there are many of waiting messages it leads to huge latency. To overcome this, this library supports onPrepare and onCommit to make a processing cycle like onPrepare -> a loop of onMessage until a queue is empty -> onCommit.

`typescript
class Adder {
private value = 0;

constructor(public readonly id: string) {}

public onPrepare = async () => {
// Load context from the outer storage.
};
public onCommit = () => {
// Store context to the outer storage.
};

public onMessage = (message: { delta: number }) => {
// Modify context in memory.
this.value += message.delta;
};
}

const env = { ...Actor.singleConsumer, ...subsys, ...new Adder(adder-1) };
Actor.send(env, {
item: { delta: 10 },
awaitPolicy: Actor.AwaitPolicy.Commit,
awaitTimeoutMillis: 1000,
})
.then(/ HAPPY /) // It would be called after onCommit.
.catch(/ SAD /);
`

$3

Sometimes, we want to use fire-and-forget producer and dedicated consumer to improve overall latency. And, in this case, _bulk-message-handler_ is better than _single-message-handler_.

`typescript
// To reduce code size, use an environment tailored to post.
await Actor.post(
{
id: adder-1,
awaiter: {
wait: subsys.awaiter.wait,
},
queue: {
push: subsys.queue.push,
},
logger: subsys.logger, // optional
},
{ item: { delta: 10 } }
)
.then(/ HAPPY /)
.catch(/ SAD /);

// Or you can use enqueue, which doesn't even need awaiter.
Actor.enqueue(
{
id: adder.id,
queue: {
push: actorSubsys.queue.push,
},
logger: actorSubsys.logger, // optional
},
{ item: { delta: 1 } }
)
.then(/ HAPPY /)
.catch(/ SAD /);
`

`typescript
class Adder {
private value = 0;

constructor(public readonly id: string) {}

// It can process multiple messages at one time.
public onMessages = (messages: { delta: number }[]) => {
for (const message of messages) {
this.value += message.delta;
}
};
}

// This bulk processor would be alive in 60 seconds.
const env = { ...Actor.bulkConsumer, ...subsys, ...new Adder(adder-1) };
Actor.tryToProcess(env, { aliveMillis: 60 * 1000 });
`

$3

Preventing to be a victim, it supports aliveMillis and shiftable when processing messages from a queue. If a timeout occurred while executing tryToProcess, it gives up to process and occurs shift event. It is useful to use in a container which has a limited lifetime such as AWS Lambda.

`typescript
const subsysWithShift = {
...subsys,
shift: async (actorId: string) => {
// Invoke a new AWS Lambda to process remaining messages in this actor.
},
};

const env = { ...Actor.singleConsumer, ...subsys, ...new Adder(adder-shift) };
Actor.send(
env,
{
item: { delta: 10 },
},
{ aliveMillis: 5 * 1000, shiftable: true } // A usual timeout of API Gateway
);
`

It can be expanded a distributed actor system easily when both of a queue and a lock work properly in shared instances.

$3

We can forget about awaiter for a while and focus on the more basic structure. They are enqueue and eventLoop. This is useful if we want to process other tasks in a loop as well as process a message when the actor's lock is occupied. For example, we could process the game logic by polling game messages from the actor queue.

`typescript
import * as Actor from "@yingyeothon/actor-system";
import * as InMemorySupport from "@yingyeothon/actor-system/lib/support/inmemory";

const subsys = {
queue: new InMemorySupport.InMemoryQueue(),
lock: new InMemorySupport.InMemoryLock(),
awaiter: new InMemorySupport.InMemoryAwaiter(),
};

class Game {
constructor(public readonly id: string) {}

public loop = async (poll: () => Promise) => {
while (this.running) {
this.processMessages(await poll());
this.tick();
}
};
}

const subsys = {
queue: new InMemorySupport.InMemoryQueue(),
lock: new InMemorySupport.InMemoryLock(),
};

async function main() {
const game = new Game("GAME_ID");
await Actor.eventLoop({
...subsys,
...game,
});
}

async function sendMessage(gameId: string, message: GameMessage) {
await Actor.enqueue(
{
...subsys,
id: gameId,
},
{ item: message }
);
}
``

License

MIT