✨Extensible worker for Node.js that works with both Zeebe and Camunda BPM platforms powered by TypeScript ✨
We needed a framework to help us quickly build workers used to execute tasks.
This package can be useful because:
- Experiment and choose the Camunda platform you want without rewritting the business logic.
- At this moment, Zeebe doesn't provide all BPMN components. Zeebe is new and some unexpected bugs can appear during development so we can easily revert back to the the former platform if an issue was to rise.
- Instead of depending directly from a Camunda client, this project provides an abstraction layer. This way it’s easier to change the client or to make your own.
- You want to have a worker standardization.
- Uniformisation. Indeed, you can use both platforms depending project needs.
- Added features like automated tracing.
- This package enforce feature parity between Zeebe and Camunda BPM through the client libraries. Some features exposed to the Camunda BPM platform are not presents in this package because we couldn't provide them if we switch to Zeebe. This limitation is to guide developers to prepare migration.
- Documentation is available in this folder
- Comprehensive API documentation is available online and in the
docs
subdirectory
Package | Description |
---|---|
workit-types | This package provides TypeScript interfaces and enums for the Workit core model. |
workit-core | This package provides default and no-op implementations of the Workit types |
Package | Description |
---|---|
workit-bpm-client | This module provides a full control over the Camunda Bpm platform. It use camunda-external-task-client-js by default. |
workit-zeebe-client | This module provides a full control over the Zeebe platform. It use zeebe-node and zeebe-elasticsearch-client by default. |
workit-camunda | This module allows you to switch between Camunda BPM and Zeebe easily. It use workit-bpm-client and workit-zeebe-client by default. |
npm i workit-camunda
or using the generator below
This generator will help you during your development with this library. It provides handy tools.
npm i -g workit-cli
workit init
workit create task --file /your/path.bpmn
workit create task
Switching between Zeebe and the bpmn platform is easy as specifying a TAG
to the IoC.
const worker = IoC.get<Worker>(CORE_IDENTIFIER.worker, TAG.camundaBpm); // or TAG.zeebe
worker.start();
worker.run();
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
const fullpath = `${process.cwd()}/sample/BPMN_DEMO.bpmn`;
await manager.deployWorkflow(fullpath);
Zeebe: You will need elasticsearch instance.
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.getWorkflows()
Zeebe: You will need elasticsearch instance.
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.getWorkflow({ bpmnProcessId: "DEMO" });
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.updateVariables({
processInstanceId: "5c50c48e-4691-11e9-8b8f-0242ac110002",
variables: { amount: 1000 }
});
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.publishMessage({
correlation: {},
name: "catching",
variables: { amount: 100 },
timeToLive: undefined, // only supported for Zeebe
messageId: "5c50c48e-4691-11e9-8b8f-0242ac110002"
});
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.createWorkflowInstance({
bpmnProcessId: "MY_BPMN_KEY",
variables: {
hello: "world"
}
});
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.cancelWorkflowInstance("4651614f-4b3c-11e9-b5b3-ee5801424400");
const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm); // or TAG.zeebe
await manager.resolveIncident("c84fce6c-518e-11e9-bd78-0242ac110003");
You can define many tasks to one worker. It will handle all messages and will route to the right tasks.
export class HelloWorldTask extends TaskBase<IMessage> {
// You can type message like IMessage<TBody, TProps> default any
public execute(message: IMessage): Promise<IMessage> {
const { properties } = message;
console.log(`Executing task: ${properties.activityId}`);
console.log(`${properties.bpmnProcessId}::${properties.processInstanceId} Servus!`);
// put your business logic here
return Promise.resolve(message);
}
}
enum LOCAL_IDENTIFIER {
// sample_activity must match the activityId in your bpmn
sample_activity= 'sample_activity'
}
// Register your task
IoC.bindTo(HelloWorldTask, LOCAL_IDENTIFIER.sample_activity);
You can even make complex binding like
IoC.bindTask(HelloWorldTaskV2, LOCAL_IDENTIFIER.activity1, { bpmnProcessId: BPMN_PROCESS_ID, version: 2 });
If you have installed workit-cli
, you can do workit create task
and everything will be done for you.
const worker = IoC.get<Worker>(CORE_IDENTIFIER.worker, TAG.zeebe); // or TAG.camundaBpm
worker.once('starting', () => {
// slack notification
});
worker.once('stopping', () => {
// close connections
});
worker.once('stopped', () => {
// slack notification
});
const handler = worker.getProcessHandler();
handler.on('message', (msg: IMessage) => {
// log/audit
});
handler.on('message-handled', (err: Error, msg: IMessage) => {
if (err) {
// something wrong
} else {
// everything is fine
}
});
worker.start();
worker.run(); // Promise
worker.stop(); // Promise
const workerConfig = {
interceptors: [
async (message: IMessage): Promise<IMessage> => {
// do something before we execute task.
return message;
}
]
};
IoC.bindToObject(workerConfig, CORE_IDENTIFIER.worker_config);
By default, we bound a NoopTracer
but you can provide your own and it must extend Tracer.We strongly recommand to use this kind of pattern in your task: Domain Probe pattern. But here an example:
// Simply bind your custom tracer object like this
IoC.bindToObject(tracer, CORE_IDENTIFIER.tracer);
export class HelloWorldTask extends TaskBase<IMessage> {
private readonly _tracer: Tracer;
constructor(tracer: Tracer) {
this._tracer = tracer
}
public async execute(message: IMessage): Promise<IMessage> {
const { properties } = message;
console.log(`Executing task: ${properties.activityId}`);
console.log(`${properties.bpmnProcessId}::${properties.processInstanceId} Servus!`);
// This call will be traced automatically
const response = await axios.get('https://jsonplaceholder.typicode.com/todos/1');
// you can also create a custom trace like this :
const currentSpan = tracer.getCurrentSpan();
const span = this._tracer.startSpan('customSpan', {
parent: currentSpan,
kind: SpanKind.CLIENT,
attributes: { key: 'value' },
});
console.log();
console.log('data:');
console.log(response.data);
// put your business logic here
// finish the span scope
span.end();
return Promise.resolve(message);
}
}
You can look to sample
folder where we provide an example (parallel.ts) using Jaeger.
See get started section with OpenTelemetry
const configBase: ICamundaConfig = {
workerId: 'demo',
baseUrl: `__undefined__`,
topicName: 'topic_demo'
};
// For Camunda BPM platform
const bpmnPlatformClientConfig = { ...configBase, baseUrl: 'http://localhost:8080/engine-rest', maxTasks: 32, autoPoll: false, use: [] };
IoC.bindToObject(bpmnPlatformClientConfig, CORE_IDENTIFIER.camunda_external_config);
// For Zeebe platform
const zeebeClientConfig = { ...configBase, baseUrl: 'localhost:2650', timeout: 2000 };
// For Zeebe exporter (Elasticsearch instance)
const zeebeElasticExporterConfig = {
url: `http://localhost:9200`,
};
IoC.bindToObject(zeebeClientConfig, CORE_IDENTIFIER.zeebe_external_config);
IoC.bindToObject(zeebeElasticExporterConfig, CORE_IDENTIFIER.zeebe_elastic_exporter_config)
By default, we define simple strategy for success or failure. We strongly recommend you to provide yours as your app trigger specific exceptions. Strategies are automatically handled. If an exeption is bubble up from the task, failure strategy is raised, otherwise it's success.
// the idea is to create your own but imagine that your worker works mainly with HTTP REST API
class ServerErrorHandler extends ErrorHandlerBase {
constructor(config: { maxRetries: number }) {
super(config);
}
public isHandled(error: IErrorResponse<IResponse<IApiError>>): boolean {
return error.response.status >= 500;
}
public handle(error: IErrorResponse<IResponse<IApiError>>, message: IMessage): Failure {
const retries = this.getRetryValue(message);
return new Failure(error.message, this.buildErrorDetails(error, message), retries, 2000 * retries);
}
}
// You got the idea...
// You could create also
// BadRequestErrorHandler
// TimeoutErrorHandler
// UnManagedErrorHandler
// ...
// Then you could build your strategy
/// "FailureStrategy" implements "IFailureStrategy", this interface is provided by workit-camunda
const strategy = new FailureStrategy([
new AxiosApiErrorHandler(errorConfig, [
new BadRequestErrorHandler(errorConfig),
new TimeoutErrorHandler(errorConfig),
new ServerErrorHandler(errorConfig),
new UnManagedErrorHandler(errorConfig),
//...
]),
new ErrorHandler(errorConfig)
]);
// worker will use your new strategy
IoC.bindToObject(strategy, CORE_IDENTIFIER.failure_strategy);
We use Jest.
npm test
- zeebe-node - nodejs client for Zeebe
- camunda-external-task-client-js - nodejs client for Camunda BPM
- inversify - Dependency injection
- opentelemetry - add instrumentation to the operations (provides a single set of APIs, libraries to capture distributed traces)
- Allow Javascript developers to write code that adheres to the SOLID principles.
- Facilitate and encourage the adherence to the best OOP and IoC practices.
- Add as little runtime overhead as possible.
TODO: provide helm chart. But in the meantime, you can do for development:
kubernetes/run
In your terminal
docker/run
docker run -d --name camunda -p 8080:8080 camunda/camunda-bpm-platform:latest
// Go: http://localhost:8080/camunda - user/password : `demo/demo`
Click to expand
- Add tests
- Improve docs
- Make sample and confirm compatibility with DMN
- Adding a common exception error codes between Manager clients
- Add metrics by using prometheus lib
We use SemVer for versioning. For the versions available, see the tags on this repository.
workit-* | Zeebe | Camunda BPM |
---|---|---|
>=4.0.5 | 0.22.1 | 7.6 to latest |
3.2.x <=4.0.4 | 0.20.x < 0.20.1 | 7.6 to latest |
3.1.x | 0.20.x < 0.20.1 | 7.6 to latest |
2.2.0 | 0.20.x < 0.20.1 | 7.6 to latest |
2.1.0 | 0.19.x | 7.6 to latest |
2.0.1 | 0.18.x | 7.6 to latest |
< 1.0.0 | <= 0.17.0 | 7.6 to latest |
See the list of contributors who participated in this project.
Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.
This project is licensed under the MIT License - see the LICENSE file for details
- Josh Wulf - zeebe-node inspired me during
workit-cli
development