michalcz / scramjet-kafka

Scramjhet

A simple module that exposes some simple methods to work with Kafka topics using {@link https://scramjet.eu/ scramjet}.

Installation:

$ npm install --save scramjet-kafka

Here's a sample case of augmenting a topic using an API: calcalutation of all transactions to a single currency in your kafka topic

  await (
      require('scramjet-kafka')
          .consume(
              "zk-kafka01.local:2021",      // Zookeeper connection string
              "transactions"                // Topic
          )
          .assign(                          // assign the calcalutation data
              async (transaction) => {
                  if (transaction.currency === "EUR") return { eur_value: transaction.value };
                  const exchange_rate = await getCurrencyExchangeRate(transaction.currency, "EUR");
                  return {
                      eur_value: exchange_rate * transaction.value
                  }
              }
          )
          .produce(                           // produce new topic with augmented data
              "zk-kafka01.example.com:2021",
              "transactionsInEUR"
          )
  );

Or an even simpler version:

await (
    require('scramjet-kafka').augment(
        "zk-kafka01.local:2021",      // Zookeeper connection string
        "transactions",               // From Topic
        (stream) => stream            // Transform the stream however you like
            .assign(async (transaction) => ({
                eur_value: await getEURValue(transaction.value, transaction.currency)
            })),
        "transactionsInEUR"           // To Topic
    )
);

Documentation

Table of Contents

• KafkaStream
  • Parameters
  • connect
  • addTopics
    • Parameters
  • removeTopics
    • Parameters
  • commit
  • setOffset
    • Parameters
• plugin
  • viaKafka
    • Parameters
  • produceKafka
    • Parameters
• sjKafka
• augment
  • Parameters
• consume
  • Parameters
  • Properties

KafkaStream

Extends DataStream

A scramjet.DataStream augmented with Kafka specific methods.

Parameters

• options
• consumerOptions Object Consumer options (optional, default {})
• streamOptions Object Options passed to scramjet

connect

Opens up connection to kafka and starts streaming.

addTopics

Add topics to the stream

Parameters

• topics Array<(String | Topic)> list of topics to listen on
• args Array additional arguments to ConsumerStream::addTopics

removeTopics

Removes topics from the stream

Parameters

• topics Array<String> list of topics to remove

commit

Commits at the current position

setOffset

Sets read offset at current position

Parameters

• topic String topic name in kafka
• partition Number where to start reading
• offset Number kafka partition number

plugin

Scramjet StringStream plugin - the following methods are added to all scramjet streams

viaKafka

Plugin to scramjet::StringStream - push to kafka and pull on the other end.

This may be used to allow burst flow above memory limits.

Parameters

• client (Client | KafkaClient) KafkaNode client to Zookeeper or direcly Kafka
• topic (String | Topic) topic - will be autogenerated if not given (optional, default null)

Returns Promise resolved when fromTopic stream ends.

produceKafka

Send the stream to the specified Kafka topic.

Parameters

• client (Client | KafkaClient) KafkaNode.client
• topic (String | Topic) topic - will be autogenerated if not given (optional, default null)

Returns Promise resolved when fromTopic stream ends.

sjKafka

Scramjet Kafka module exports

Type: Object

augment

Fetches a stream from Kafka topic performs declared operations and publishes to the another topic.

The transforms as in scramjet can be asynchronous, synchronous, even multi-threaded.

{@see https://scramjet.eu/ Scramjet documentation}

Parameters

• client (Client | KafkaClient) KafkaNode client to Zookeeper or direcly Kafka
• fromTopic (String | Topic) topic to consume
• use UseCallback transforms callback or scramjet module
• toTopic (String | Topic) topic to produce3

Returns Promise resolved when fromTopic stream ends.

consume

Consume a topic from kafka and return a new KafkaStream

Parameters

• client (Client | KafkaClient) KafkaNode client to Zookeeper or direcly Kafka
• topics Array<(String | Topic)> Topics to pull from Kafka (optional, default [])

Properties

• topic String topic name in kafka
• offset Number where to start reading
• partition Number kafka partition number

License

MIT - see LICENSE

scramjet-kafka's Issues

Kinda works with Kafka only on localhost

Seems that not everywhere the options are passed just right...