@betsys-nestjs/cluster-worker
v4.0.0
Published
This library provides a framework to handle multi-process applications. It can spawn, manage and destroy workers from master process.
Downloads
26
Maintainers
Keywords
Readme
Cluster - worker library
This library is quite helpful to manage multi-process applications with slave master architecture. Take this readme as a manual, how to write such an application.
Environment variables
There are no variables necessary for this library.
Dependencies
| Package | Version | | --------------------- | ------- | | on-headers | ^1.0.2 | | @nestjs/common | ^10.0.0 | | @nestjs/core | ^10.0.0 | | @nestjs/event-emitter | ^2.0.0 | | @nestjs/schedule | ^3.0.0 | | child-process | ^1.0.0 | | reflect-metadata | ^0.1.13 | | rxjs | ^7.1.0 |
Usage
In order to use this module, you need to create 2 modules - one with cluster logic (master) and one with worker logic (slave) and then incorporate the cluster logic module into the app module.
Cluster module
import { ClusterModule } from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import { CallbackProvider } from './listener/callback.provider';
import { ReplicationWorkerProvider } from './provider/replication-worker.provider';
@Module({
imports: [
ClusterModule.forRoot(),
],
providers: [
CallbackProvider,
ReplicationWorkerProvider,
],
exports: [
ReplicationWorkerProvider,
],
})
export class ClusterLogicModule {
}
ClusterModule
is the module imported from the library. CallbackProvider
is a class defining, how the cluster should react to messages from worker.
import { Injectable } from '@nestjs/common';
enum MyMessage {
DoSomething = 'DoSomething',
}
@Injectable()
export class CallbackProvider implements CallbackProviderInterface<MyMessage> {
private callbacks: { message: MyMessage; callable: ClusterCallbackType }[] = [];
constructor() {
this.addCallback(MyMessage.DoSomething, this.doSomething);
}
doSomething(): void {
...
}
}
When you already have defined behavior of master for given message, you need to implement a worker provider:
import {
SingleWorkerProvider,
WorkerProviderService,
WorkerStatusWaitingService,
InjectLogger,
ClusterWorkerLoggerInterface,
} from '@betsys/cluster-worker';
import { forwardRef, Inject, Injectable } from '@nestjs/common';
import { CallbackProvider } from '../listener/callback.provider';
enum MyWorkerProviderEnum {
SillyWorker = 'SillyWorker',
}
enum MyMessage {
DoSomething = 'DoSomething',
}
@Injectable()
export class MyWorkerProviderService extends SingleWorkerProvider<never, MyMessage, MyWorkerProviderEnum> {
constructor(
@InjectLogger()
protected readonly logger: ClusterWorkerLoggerInterface,
protected readonly workerProvider: WorkerProviderService,
@Inject(forwardRef(() => CallbackProvider))
protected readonly callbackProvider: CallbackProvider,
protected readonly waitingService: WorkerStatusWaitingService,
) {
super(
logger,
workerProvider,
callbackProvider,
MyWorkerProviderEnum.SillyWorker,
'spawn message that appears in the log',
waitingService,
);
}
}
Such a worker provider can extend SingleWorkerProvider
(when you need
just a single worker for given task) or MultiWorkerProvider
(when you
need a pool of workers for the same task).
A slave worker process is represented as a WorkerMeta
object in the cluster:
import cluster from 'cluster';
import { WorkerStatus } from './worker-status.enum';
export interface WorkerMeta<TData = any, TWorkerStatus = any> {
data?: TData;
process: cluster.Worker;
status: WorkerStatus | TWorkerStatus;
pingPending: boolean;
}
With the type TData
you can define additional data for the worker
(i.e. if it has to consume some queue from RabbitMQ, you can define
the queue name over here) and TWorkerStatus
defines the status
the worker is in (it is your custom worker status, i.e. Replicating
).
WorkerStatus
has three values: Booting
, Ready
and Error
.
MultiWorkerProvider
offers following methods:
// Respawns all workers, which WorkerStatus is Error
respawnWorkersWithError(): Promise<WorkerMeta<TWorkerData>[]>;
// Spawns workers with data you want to pass to it
spawnWorkersWithData(
workerData: TWorkerData[],
mapper: (data: TWorkerData) => WorkerMeta<TWorkerData>,
): Promise<WorkerMeta<TWorkerData>[]>;
// Spawn workers without data
spawnWorkers(count: number): Promise<WorkerMeta<TWorkerData>[]>;
// Sends a message to all workers
sendToAllWorkers(message: string, sendHandle?: string, callback?:
(error: Error | null) => void): boolean;
// This is obvious, right?
getWorkers(): WorkerMeta<TWorkerData>[];
An example of mapper to spawn workers with data would be:
(data: MyDataType): WorkerMeta<MyDataType> => ({
data,
pingPending: false,
process: cluster.fork(),
status: WorkerStatus.Booting,
});
Number of workers will be defined by number of elements in workerData
parameter.
SingleWorkerProvider
provides following methods:
spawnWorker(): Promise<WorkerMeta<TWorkerData>>;
getWorker(): WorkerMeta<TWorkerData> | null;
// Little bit morbid, right?
destroyWorker(): Promise<void>;
Worker module
When you have everything set up with cluster, it comes the time to tune up the worker:
import {
ClusterModule,
StatusService,
WorkerListenerService,
WorkerModule as LibWorkerModule,
WorkerStatus,
WorkerCallbackProvider,
InjectLogger,
ClusterWorkerLoggerInterface,
} from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import cluster from 'cluster';
@Module({
imports: [
ClusterModule.forRoot(),
LibWorkerModule.forRoot(),
],
providers: [
WorkerCallbackProvider,
],
})
export class WorkerLogicModule {
constructor(
@InjectLogger()
private readonly logger: ClusterWorkerLoggerInterface,
private readonly statusService: StatusService,
private readonly workerCallbackProvider: WorkerCallbackProvider,
private readonly workerListenerService: WorkerListenerService,
) {
if (!cluster.isWorker) {
throw new Error('This service must be run only from worker process');
}
}
async onApplicationBootstrap(): Promise<void> {
const callbacks = this.workerCallbackProvider.getCallbacks();
this.workerListenerService.addCustomListeners(callbacks);
this.statusService.changeStatus(WorkerStatus.Ready);
}
}
In WorkerCallbackProvider
you can define, how the worker will react
to a message from master.
enum MyMessage {
DoSomethingSlave = 'DoSomethingSlave',
}
@Injectable()
export class WorkerCallbackProvider
implements WorkerCallbackProviderInterface<MyMessage> {
private callbacks: { message: MyMessage; callable: WorkerCallbackType }[] = [];
constructor() {
this.doSomething = this.doSomething.bind(this);
this.addCallback(MyMessage.DoSomethingSlave, this.doSomething);
}
doSomething(): void {
...
}
}
AppModule and starting of the app
Now it comes the time to put everything together in AppModule
:
import { ClusterModule } from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import { ClusterLogicModule } from './cluster-logic/cluster-logic.module';
@Module({
imports: [
ClusterLogicModule,
ClusterModule.forRoot(),
],
})
export class AppModule implements BeforeApplicationShutdown {
constructor(
// This is from cluster-logic module
private myWorkerProvider: MyWorkerProvider,
private readonly workerStatusWaitingService: WorkerStatusWaitingService,
) {
}
async onApplicationBootstrap(): Promise<void> {
const worker = await this.myWorkerProvider.spawnWorker();
await this.workerStatusWaitingService.waitUntilWorkerMetaHasStatus(
worker,
WorkerStatus.Ready,
);
worker.process.send({ type: MyMessage.DoSomething });
}
}
Now all you have to do is to define main.ts
:
import cluster from 'cluster';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { WorkerModule } from './worker-logic/worker-logic.module';
async function bootstrapMaster(): Promise<void> {
const app = await NestFactory.create(AppModule);
app.enableShutdownHooks();
const port = Number(process.env.CONFIG_LISTEN_PORT);
await app.listen(port);
const logger = await app.resolve(Logger);
logger.info(`Application listens on port ${port}`);
}
async function bootstrapWorker(): Promise<void> {
const app = await NestFactory.create(WorkerLogicModule);
app.enableShutdownHooks();
await app.init();
}
const bootstrap = cluster.isMaster ? bootstrapMaster : bootstrapWorker;
bootstrap()
.catch((e) => {
console.error(e);
process.exit(1);
});
So basically you start the AppModule
, which is the master process
(so bootstrapMaster
is called). Within this process, a new slave
process is spawned and the main.ts
is called again. However,
this time, as it is a slave process, bootstrapWorker
is called.
Now you can send messages from master to worker or from worker to master
(you just need to define callbacks for a particular message).