index.js

var Worker = require('camunda-worker-node');

var Backoff = require('camunda-worker-node/lib/backoff');
var Metrics = require('camunda-worker-node/lib/metrics');

var engineEndpoint = process.env.ENGINE_URL || 'http://localhost:8080/engine-rest';

var uuid = require('uuid');


var debugWorker = require('debug')('orderProcess:worker');

var debugShipment = require('debug')('orderProcess:worker:shipment');
var debugCheckout = require('debug')('orderProcess:worker:checkout');

var worker = new Worker(engineEndpoint, {
  maxTasks: 10,
  use: [
    [ Backoff, { maxActiveTasks: 50 } ],
    Metrics
  ]
});


async function shipOrder(context) {

  const {
    variables
  } = context;

  var order = variables.order;

  debugShipment('processing order[id=%s]', order.orderId);

  await delay(Math.trunc(Math.random() * 1.5));

  if (Math.random() > 0.8) {
    debugShipment('failed to ship order[id=%s]', order.orderId);

    throw new Error('failed to process shipment: RANDOM STUFF');
  }

  // do actual work here, write database, provision goods
  order.shipmentId = uuid.v4();
  order.shipped = true;

  debugShipment('shipping order[id=%s] with shipmentId=%s', order.orderId, order.shipmentId);

  // notify we are done with an updated order variable
  return {
    variables: {
      order: order
    }
  };
}

async function checkout(context) {

  const {
    variables,
    extendLock
  } = context;

  const goods = variables.goods;

  if (!goods || goods.length === 0) {
    throw new Error('no goods in basket');
  }

  // do actual work here, write database, reserve goods
  const order = {
    orderId: uuid.v4(),
    goods
  };

  debugCheckout(
    'created order[orderId=%s] with %s goods',
    order.orderId,
    order.goods.length
  );

  if (Math.random() > 0.6) {
    debugCheckout('delayed order[orderId=%s]', order.orderId);

    await extendLock(5000);

    await delay(Math.trunc(Math.random() * 3));
  }

  // notify we are done with a new order variable
  return {
    variables: {
      order: order
    }
  };
}

worker.subscribe('orderProcess:shipment', [ 'order' ], shipOrder);

worker.subscribe('orderProcess:checkout', [ 'goods' ], checkout);


worker.on('start', function() {
  debugWorker('starting');
});

worker.on('poll', function() {
  debugWorker('polling');
});

// handle worker errors
worker.on('error', function(err) {
  debugWorker('error: %s', err);
});


// helpers ////////////////////

function delay(seconds) {

  return new Promise(function(resolve) {
    setTimeout(resolve, seconds * 1000);
  });

}