WorkIt

Français

✨Extensible worker for Node.js that works with both AWS Step function and Camunda BPM platforms powered by TypeScript ✨

Motivation

We needed a framework to help us quickly build workers used to execute tasks.

This package can be useful because:

• Experiment and choose the platform you want without rewritting the business logic. Today, only Camunda and AWS Step function clients are maintained
• 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.

Quickstart

Get started in 2 minutes.

Documentation

• Documentation is available in this folder
• Comprehensive API documentation is available online and in the docs subdirectory
• Examples

API

Implementation / Clients

Installing

npm i @villedemontreal/workit

How to use

Switching between platforms is easy as specifying a TAG to the IoC.

Run worker

const worker = IoC.get<Worker>(CORE_IDENTIFIER.worker, TAG.camundaBpm);

worker.start();
worker.run();

Deploy a workflow

const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm);
const fullpath = `${process.cwd()}/sample/BPMN_DEMO.bpmn`;
await manager.deployWorkflow(fullpath);

Get workflows

const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm);
await manager.getWorkflows()

Get a workflow

const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm);
await manager.getWorkflow({ bpmnProcessId: "DEMO" });

Create workflow instance

const manager = IoC.get<IWorkflowClient>(CORE_IDENTIFIER.client_manager, TAG.camundaBpm);
await manager.createWorkflowInstance({
    bpmnProcessId: "MY_BPMN_KEY",
    variables: {
        hello: "world"
    }
});

Define tasks (your bpmn activities)

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.

Worker life cycle and events

const worker = IoC.get<Worker>(CORE_IDENTIFIER.worker, 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

Interceptors

const workerConfig = {
  interceptors: [
    async (message: IMessage): Promise<IMessage> => {
      // do something before we execute task.
      return message;
    }
  ]
};

IoC.bindToObject(workerConfig, CORE_IDENTIFIER.worker_config);

OpenTelemetry

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

Define your config for the platform you want to use

TODO show for step function

Define your strategies in case of failure or success

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
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);

Running the tests

We use Jest.

npm test

Built With

• camunda-external-task-client-js - nodejs client for Camunda BPM
• @aws-sdk/client-sqs - nodejs client for receiving messages from the queue
• @aws-sdk/client-sfn - nodejs client for managing state machines and acknowledging process
• inversify - dependency injection
• opentelemetry - add instrumentation to the operations (provides a single set of APIs, libraries to capture distributed traces)

Philosophy

1. Allow Javascript developers to write code that adheres to the SOLID principles.
2. Facilitate and encourage the adherence to the best OOP and IoC practices.
3. Add as little runtime overhead as possible.

Docker

Bpmn platform

docker run -d --name camunda -p 8080:8080 camunda/camunda-bpm-platform:latest
// Go: http://localhost:8080/camunda - user/password : `demo/demo`

More details

TODO

Versionning

We use SemVer for versioning. For the versions available, see the tags on this repository.

Maintainers

See the list of contributors who participated in this project.

Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

License

This project is licensed under the MIT License - see the LICENSE file for details