@level-two/kafka

LevelTwo - Kafka

A message broker extension for LevelTwo. Refer to the Worker API for function definitions.

Usage

import { createLevelTwoKafka } from "@level-two/kafka";

// Integrates RabbitMQ message broker into a LevelTwo instance
const levelTwo = createLevelTwoKafka({
  topic: "level-two-broadcast",
  clientOptions: {
    brokers: ["localhost:9092"],
  },
});

// Create a worker for getting cacheable customer objects
const worker = levelTwo.createWorker("customer", async (ids, earlyWrite) => {
  const rows = await mysql.query(
    "SELECT id, name FROM customers WHERE id IN (?)",
    [ids]
  );

  rows.forEach((row) => earlyWrite(row.id, row));
});

// Service method for getting a single customer
export async function getCustomer(id: string): Promise<Customer> {
  return await worker.get(id);
}

// Service method for getting a list of customers
export async function getCustomerList(ids: string[]): Promise<Customer[]> {
  return await worker.getMulti(ids);
}

Settings

topic

Topic name for sending/receiving actions

clientOptions

Kafka client options

consumerOptions

Consumer connection options

producerOptions

Producer connection options

remoteCache

LevelTwo remote cache extension

cacheDefaults

LevelTwo worker cache defaults

Changelog

v2.0.0

• #11 Upgrading all dependencies
• #10 skipRemoteCache: Configure workers to skip remote-cache binding
• #9 getEntry & getEntryMulti: Fetching meta information along with a cached values
• #8 SingleKeyWorker: Cache worker for a single key entry

Breaking Changes

Starting with version 2.0.0: Iterable, peek, peekMulti, values, entries, forEach all return CachedEntry wrapped result values instead of the ResultValue directly.

Switching to a wrapped CachedEntry will allow for current and future expansion on utilities for metric tracking of entries.