rxjs-kafkajs

rxjs + kafkajs

Usage

import { RxKafka } from 'rxjs-kafkajs';

const rxKafka = new RxKafka({
	brokers: [
		`localhost:9092`,
	],
});

const observable = rxKafka.consumeMergeMapProduce({
	groupId: 'group',
	transactionalId: 'transactional-id-1',
	topic: 'topic-1',
	async * project({ message }) {
		if (message.value?.toString() === 'foo') {
			yield {
				topicMessages: [
					{
						topic: 'topic-2',
						messages: [
							{
								value: 'foo1',
							},
							{
								value: 'foo2',
							},
						],
					},
				],
			};
		}
	},
});

await Rx.firstValueFrom(observable);