hiot-kafka

hiot-app middleware with kafka

handleEnvelope

The hiot-kafka package contains a handleEnvelope helper which is there to handle the administrative tasks of handling a kafka message (i.e. logging, acking/rejecting envelopes, handling exception tracking). To use it you need to pass it:

• A function which handles the kafka message (a function which returns a promise)
• The kafka broker
• The application's logger
• A function which handles exceptions

broker.consume(
  "things.values",
  handleEnvelope(thingValues, broker, logger, onUncaughtException)
);

Then an example handler function would look like this:

const thingValues = async (message, broker) => {
  if (!validator.thingValues(message)) {
    return { status: "ignored", reason: "invalid message" };
  }

  let thingId = message.value.thingId;
  thing = await thingsAPI.fetchThing(thingId);

  if (!thing) {
    return { status: "ignored", reason: "could not find thing" };
  }

  let oldState = {};
  if (thing && thing.state) {
    oldState = thing.state;
  }

  let newState = {
    state: {
      humidity: message.value.values[0].humidity
        ? message.value.values[0].humidity
        : oldState.humidity,
      temperature: message.value.values[0].temperature
        ? message.value.values[0].temperature
        : oldState.temperature,
      battery: message.value.values[0].battery
        ? message.value.values[0].battery
        : oldState.battery,
      connection: "online",
    },
  };

  await thingsAPI.patch(thingId, message.value.correlationId, newState);

  return { status: "handled", thingId };
};

This will produce standard log messages like these to indicate when handling of a message is started and when it is completed (either with a status of ack or a status of reject)

The logs look like this (prettified for easier reading):

{
  "name": "things-svc",
  "hostname": "things-svc-d8b68cb86-s57hb",
  "pid": 1,
  "level": 30,
  "label": "hiot-kafka-handle-envelope",
  "topic": "things.values",
  "msg": "Received message for topic: things.values",
  "time": "2021-04-01T19:48:44.491Z",
  "v": 0
}
{
  "name": "things-svc",
  "hostname": "things-svc-d8b68cb86-s57hb",
  "pid": 1,
  "level": 30,
  "label": "hiot-kafka-handle-envelope",
  "topic": "things.values",
  "status": "ack",
  "metadata": {
    "status": "handled",
    "thingId": "9af5fbc0708711ebb39301f66685f25d"
  },
  "msg": "Received success for things.values",
  "time": "2021-04-01T19:48:44.521Z",
  "v": 0
}

Since the handling function returns a promise, you need to either reject the promise (for the async function above an error is raised) or resolve it (the return statements). When resolving the promise you can optionally return an object which will be put into the metadata key of the logs. This can be used to log a status key (i.e. ignored or handled) and other details which are particularly important to debugging the message (i.e. a thingId or a userId)

A note on ignoring vs. failing

If a service gets a message which it isn't programmed to handled, it is recommended that the message be acknowledged rather that rejected. Kafka is currently configured to simply drop messages when they are rejected, if we ever wanted to implement retries in kafka we should simply handle/acknowledge messages that we want to ignore. Otherwise, if we do implement retries then the message will be retried continually for no reason.

Prettier & ESlint

we use prettier to format our code. Our recommendation is to configure your editor to autoformat during save. If you open up the VS Code User's settings/preferences as UI, search for "Format On Save" and make sure to activate it. Afterward, the file should format automatically once you save it. Now you don’t need to worry about your code formatting anymore, because Prettier takes care of it.

#manual checking and apply
npx prettier --check .
npx prettier --write .

#validate passes eslint
yarn run eslint .
yarn run eslint . --fix
# or
npm run lint
npm run lint-fix