Kafka Projection Updates Configuration

Overview

The Kafka Projection Updates is a JSON file which describes for each projection a Projection Updates topic and a Strategy to generate the projection changes identifier. This file is typically used by:

• the Real-Time Updater, to generate a pr-update message;
• the Single View Trigger Generator, to execute a strategy starting from a pr-update message.

Configuration Properties

The Kafka Projection Updates is a JSON object where each key represents the name of a projection. Each entry has the following fields:

• updatesTopic: The topic where the Projection Updates messages are sent, if you haven't created a topic for the projection yet take a look to our naming conventions for the Projection Updates topics.
• strategy: the type of strategies that you want to use onto this projection. You can choose either __automatic__ or __fromFile__.

There will be different behaviors based on the strategy value. The JSON configuration can contain both automatic and manual strategies.

Automatic

kafkaProjectionUpdates.json

{
  "PROJECTION_NAME": {
    "updatesTopic": "PROJECTION_UPDATE_TOPIC",
    "strategy": "__automatic__"
  }
}

The automatic strategy handles the generation of the projection changes identifier accordingly to the paths in the Projection Changes Schema config map containing the projection.

Manual

kafkaProjectionUpdates.json

{
  "PROJECTION_NAME": {
    "updatesTopic": "PROJECTION_UPDATE_TOPIC",
    "strategy": "__fromFile__[customFunction.js]"
  }
}

The __fromFile__ keyword lets you specify a Javascript module inside the squared brackets, which will execute the a manual Strategy on the Projection Updates message.

info

The file should be mountend into a specific config map of the micro-service. Please refer to Real-Time Updater and Single View Trigger Generator configuration pages.

This Javascript file should export a default async generator function with the following parameters:

• strategyContext: object made of two properties:
  • logger: Pino logger
  • dbMongo: the configured MongoDB Database instance where the projections are stored
• updateEvent: The Projection Updates message.

The function yields one or more identifier retrieved from the update event.

Examples

Here's a couple of examples example of what a custom strategy could look like:

Simple Custom Strategy

// note: this has to be an AsyncGenerator
module.exports = async function* myCustomStrategy ({ logger, dbMongo }, { after }) {
  yield { IDENTIFIER_FIELD: after.FIELD }
}

Custom Strategy with 1:n lookup

module.exports = async function* myCustomStrategy ({ logger, dbMongo }, { after }) {
  const lookupRecords = await dbMongo.collection('lookup-collection').find({ field: after.FIELD })
  for await(const lookup of lookupRecords) {
    yield { IDENTIFIER_FIELD: lookup.FIELD } 
  }
}