Repository files navigation

effect-kafka is a Kafka client forEffect. It provides a purely functional interface to the Kafka client and integrates effortlessly with Effect ecosystem.

Effect Website:https://www.effect.website
Effect API Reference:https://effect-ts.github.io/effect
Effect Kafka Website:https://effect-kafka.website
Effect Kafka API Reference:https://floydspace.github.io/effect-kafka

Choose your preferred package manager and run one of the following commands in your terminal:

• Using npm:

  npm install effect-kafka

• Using pnpm:

  pnpm add effect-kafka

• Using yarn:

  yarn add effect-kafka

Next install one of kafka engine packages:

• KafkaJS - Fully JavaScript implementation.
• @confluentinc/kafka-javascript - JavaScript interface for C++ librdkafka implementation, which is more performant, but requires native bindings.
• @platformatic/kafka - New pure JavaScript Kafka client implementation with ambition to replace KafkaJS

Note: You can use any of the above Kafka engine packages, depending on your preference.

Let's write a simple Kafka producer and consumer usingeffect-kafka. Before everything, we need a running instance of Kafka. We can do that by saving the following docker-compose script in thedocker-compose.yml file and rundocker-compose up:

version:'2'services:zookeeper:image:confluentinc/cp-zookeeper:latestenvironment:ZOOKEEPER_CLIENT_PORT:2181ZOOKEEPER_TICK_TIME:2000ports:      -22181:2181kafka:image:confluentinc/cp-kafka:latestdepends_on:      -zookeeperports:      -29092:29092environment:KAFKA_BROKER_ID:1KAFKA_ZOOKEEPER_CONNECT:zookeeper:2181KAFKA_ADVERTISED_LISTENERS:PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME:PLAINTEXTKAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR:1

Now, we can run oureffect-kafka application:

import{NodeRuntime}from"@effect/platform-node";import{Console,Effect,Layer,Random,Schedule,Stream}from"effect";import{Consumer,Producer}from"effect-kafka";import{ConfluentKafkaJS}from"effect-kafka/ConfluentKafka";constproducer=Stream.repeatEffect(Random.nextInt).pipe(Stream.schedule(Schedule.fixed("2 seconds")),Stream.flatMap((random)=>Producer.send({topic:"random",messages:[{key:String(random%4),value:random.toString()}],}),),);constconsumer=Consumer.serveStream("random").pipe(Stream.tap((record)=>Console.log(record.value?.toString())));constprogram=Stream.merge(producer,consumer).pipe(Stream.runDrain);constProducerLive=Producer.layer({allowAutoTopicCreation:true});constConsumerLive=Consumer.layer({groupId:"group"});constKafkaLive=ConfluentKafkaJS.layer({brokers:["localhost:29092"]});constMainLive=program.pipe(Effect.provide(Layer.merge(ProducerLive,ConsumerLive)),Effect.provide(KafkaLive));NodeRuntime.runMain(MainLive);

See more examples in theexamples directory.

• Consumer
• Producer
• Consumer Streams
• Producer Streams
• Acknowledge management
• Transactions

Thank you for considering contributing to our project! Here are some guidelines to help you get started:

If you have found a bug, please open an issue on ourissue tracker and provide as much detail as possible. This should include:

• A clear and concise description of the problem
• Steps to reproduce the problem
• The expected behavior
• The actual behavior
• Any relevant error messages or logs

If you have an idea for an enhancement or a new feature, please open an issue on ourissue tracker and provide as much detail as possible. This should include:

• A clear and concise description of the enhancement or feature
• Any potential benefits or use cases
• Any potential drawbacks or trade-offs

We welcome contributions via pull requests! Here are some guidelines to help you get started:

1. Fork the repository and clone it to your local machine.
2. Create a new branch for your changes:git checkout -b my-new-feature
3. Install dependencies:pnpm install (pnpm@9.x, usingcorepack)
   • if you introduce new dependencies, please use.projenrc.ts to add them
   • then runpnpm default to update the project
4. Make your changes and add tests if applicable.
5. Run the tests:pnpm test
6. Commit your changes:git commit -am 'Add some feature'
7. Push your changes to your fork:git push origin my-new-feature
8. Open a pull request against ourmain branch.

• Please make sure your changes are consistent with the project's existing style and conventions.
• Please write clear commit messages and include a summary of your changes in the pull request description.
• Please make sure all tests pass and add new tests as necessary.
• If your change requires documentation, please update the relevant documentation.
• Please be patient! We will do our best to review your pull request as soon as possible.

By contributing to this project, you agree that your contributions will be licensed under the project'sMIT License.

We are grateful to the following sponsors for supporting this project:

Superwall