fix: improve consumer layers usage

.changeset/dull-hats-play.md

@@ -0,0 +1,5 @@
+---
+"effect-kafka": patch
+---
+
+fix consumer layers usage

README.md

@@ -68,7 +68,7 @@ services:
 Now, we can run our `effect-kafka` application:
 ```typescript
 import { NodeRuntime } from "@effect/platform-node";
-import { Console, Effect, Random, Schedule, Stream } from "effect";
+import { Console, Effect, Layer, Random, Schedule, Stream } from "effect";
 import { ConfluentKafkaJSInstance, Consumer, Producer } from "effect-kafka";
 
 const producer = Stream.repeatEffect(Random.nextInt).pipe(
@@ -83,15 +83,20 @@ const producer = Stream.repeatEffect(Random.nextInt).pipe(
   ),
 );
 
-const consumer = Consumer.serveStream("random", { groupId: "group" }).pipe(
-  Stream.tap((record) => Console.log(record.value?.toString())),
+const consumer = Consumer.serveStream("random").pipe(
+  Stream.tap((record) => Console.log(record.value?.toString()))
 );
 
 const program = Stream.merge(producer, consumer).pipe(Stream.runDrain);
 
 const ProducerLive = Producer.layer({ allowAutoTopicCreation: true });
+const ConsumerLive = Consumer.layer({ groupId: "group" });
+
 const KafkaLive = ConfluentKafkaJSInstance.layer({ brokers: ["localhost:29092"] });
-const MainLive = Effect.scoped(program).pipe(Effect.provide(ProducerLive), Effect.provide(KafkaLive));
+const MainLive = program.pipe(
+  Effect.provide(Layer.merge(ProducerLive, ConsumerLive)),
+  Effect.provide(KafkaLive)
+);
 
 NodeRuntime.runMain(MainLive);
 ```

examples/streamConsumer.ts

@@ -2,7 +2,7 @@ import { NodeRuntime } from "@effect/platform-node";
 import { Console, Effect, Stream } from "effect";
 import { Consumer, KafkaJSInstance } from "../src";
 
-const program = Consumer.serveStream("test-topic", { groupId: "group" }).pipe(
+const program = Consumer.serveStream("test-topic").pipe(
   Stream.runForEach(({ topic, partition, ...message }) =>
     Console.log({
       topic,
@@ -13,7 +13,8 @@ const program = Consumer.serveStream("test-topic", { groupId: "group" }).pipe(
   ),
 );
 
+const ConsumerLive = Consumer.layer({ groupId: "group" });
 const KafkaLive = KafkaJSInstance.layer({ brokers: ["localhost:19092"] });
-const MainLive = Effect.scoped(program).pipe(Effect.provide(KafkaLive));
+const MainLive = Effect.scoped(program).pipe(Effect.provide(ConsumerLive), Effect.provide(KafkaLive));
 
 NodeRuntime.runMain(MainLive);

examples/streamProducer.ts

@@ -1,5 +1,5 @@
 import { NodeRuntime } from "@effect/platform-node";
-import { Clock, Console, Effect, Schedule, Stream } from "effect";
+import { Clock, Console, Effect, Layer, Schedule, Stream } from "effect";
 import { ConfluentKafkaJSInstance, Consumer, Producer } from "../src";
 
 const p = Stream.repeatEffect(Clock.currentTimeMillis).pipe(
@@ -14,7 +14,7 @@ const p = Stream.repeatEffect(Clock.currentTimeMillis).pipe(
   ),
 );
 
-const c = Consumer.serveStream("test-topic", { groupId: "group" }).pipe(
+const c = Consumer.serveStream("test-topic").pipe(
   Stream.tap(({ topic, partition, ...message }) =>
     Console.log({
       topic,
@@ -29,7 +29,9 @@ const c = Consumer.serveStream("test-topic", { groupId: "group" }).pipe(
 const program = Stream.merge(p, c).pipe(Stream.runDrain);
 
 const ProducerLive = Producer.layer({ allowAutoTopicCreation: true });
+const ConsumerLive = Consumer.layer({ groupId: "group" });
+
 const KafkaLive = ConfluentKafkaJSInstance.layer({ brokers: ["localhost:19092"] });
-const MainLive = Effect.scoped(program.pipe(Effect.provide(ProducerLive))).pipe(Effect.provide(KafkaLive));
+const MainLive = program.pipe(Effect.provide(Layer.merge(ProducerLive, ConsumerLive)), Effect.provide(KafkaLive));
 
 NodeRuntime.runMain(MainLive);

examples/streams.ts

@@ -14,11 +14,7 @@ class KafkaStreamConsumerTopic extends Context.Tag("@superwall/open-revenue-tran
 const makeKafkaStreamConsumer = Effect.gen(function* () {
   const { topic, groupId } = yield* KafkaStreamConsumerTopic;
 
-  const stream = Consumer.serveStream(topic, {
-    autoCommit: false,
-    groupId,
-    fromBeginning: true,
-  }).pipe(
+  const stream = Consumer.serveStream(topic).pipe(
     Stream.zipWithPrevious,
     Stream.mapEffect(([previous, current]) =>
       Effect.gen(function* () {
@@ -38,6 +34,13 @@ const makeKafkaStreamConsumer = Effect.gen(function* () {
         return current;
       }),
     ),
+    Stream.provideSomeLayer(
+      Consumer.layer({
+        autoCommit: false,
+        groupId,
+        fromBeginning: true,
+      }),
+    ),
   );
   return {
     stream,

src/Consumer.ts

@@ -64,6 +64,15 @@ export const make: (options: {
   ) => Stream.Stream<ConsumerRecord.ConsumerRecord, never, Scope.Scope>;
 }) => Consumer = internal.make;
 
+/**
+ * @since 0.3.1
+ * @category constructors
+ */
+export const makeConsumer: (
+  options: Consumer.ConsumerOptions,
+) => Effect.Effect<Consumer, Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =
+  internal.makeConsumer;
+
 /**
  * @since 0.1.0
  * @category accessors
@@ -134,9 +143,11 @@ export const serveEffect: {
  */
 export const serveStream: (
   path: MessageRouter.Route.Path,
-  options: Consumer.ConsumerOptions,
-) => Stream.Stream<
-  ConsumerRecord.ConsumerRecord,
-  Error.ConnectionException,
-  KafkaInstance.KafkaInstance | Scope.Scope
-> = internal.serveStream;
+) => Stream.Stream<ConsumerRecord.ConsumerRecord, Error.ConnectionException, Consumer | Scope.Scope> =
+  internal.serveStream;
+
+/**
+ * @since 0.3.1
+ * @category layers
+ */
+export const layer = (options: Consumer.ConsumerOptions) => Layer.scoped(Consumer, makeConsumer(options));

src/Producer.ts

@@ -98,15 +98,6 @@ export const make: (options: {
   readonly sendBatch: (batch: Producer.ProducerBatch) => Effect.Effect<Producer.RecordMetadata[]>;
 }) => Producer = internal.make;
 
-/**
- * @since 0.2.0
- * @category accessors
- */
-export const send: (
-  record: Producer.ProducerRecord,
-) => Effect.Effect<Producer.RecordMetadata[], Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =
-  internal.send;
-
 /**
  * @since 0.2.0
  * @category constructors
@@ -116,6 +107,15 @@ export const makeProducer: (
 ) => Effect.Effect<Producer, Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =
   internal.makeProducer;
 
+/**
+ * @since 0.2.0
+ * @category accessors
+ */
+export const send: (
+  record: Producer.ProducerRecord,
+) => Effect.Effect<Producer.RecordMetadata[], Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =
+  internal.send;
+
 /**
  * @since 0.2.0
  * @category layers

src/internal/consumer.ts

@@ -24,6 +24,15 @@ export const make = (options: {
   ) => Stream.Stream<ConsumerRecord.ConsumerRecord, never, Scope.Scope>;
 }): Consumer.Consumer => Object.assign(Object.create(consumerProto), options);
 
+/** @internal */
+export const makeConsumer = (
+  options: Consumer.Consumer.ConsumerOptions,
+): Effect.Effect<Consumer.Consumer, Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =>
+  Effect.gen(function* () {
+    const instance = yield* KafkaInstance.KafkaInstance;
+    return yield* instance.consumer(options);
+  });
+
 /** @internal */
 export const serve = dual<
   {
@@ -109,12 +118,8 @@ export const serveEffect = dual<
 /** @internal */
 export const serveStream = (
   path: MessageRouter.Route.Path,
-  options: Consumer.Consumer.ConsumerOptions,
-): Stream.Stream<ConsumerRecord.ConsumerRecord, Error.ConnectionException, KafkaInstance.KafkaInstance | Scope.Scope> =>
-  Effect.gen(function* () {
-    const instance = yield* KafkaInstance.KafkaInstance;
-    return yield* instance.consumer(options);
-  }).pipe(
+): Stream.Stream<ConsumerRecord.ConsumerRecord, Error.ConnectionException, Consumer.Consumer | Scope.Scope> =>
+  consumerTag.pipe(
     Effect.map((consumer) => consumer.runStream(path)),
     Stream.flatten(),
   );