diff --git a/.changeset/rename-publisher-subscriber-to-producer-consumer.md b/.changeset/rename-publisher-subscriber-to-producer-consumer.md new file mode 100644 index 0000000..cb13496 --- /dev/null +++ b/.changeset/rename-publisher-subscriber-to-producer-consumer.md @@ -0,0 +1,73 @@ +--- +"@effect-messaging/core": major +"@effect-messaging/amqp": major +"@effect-messaging/nats": major +--- + +Rename Publisher/Subscriber to Producer/Consumer across all packages for industry-standard naming + +### Breaking Changes + +**@effect-messaging/core:** +- `Publisher` renamed to `Producer` +- `Subscriber` renamed to `Consumer` +- `PublisherError` renamed to `ProducerError` +- `SubscriberError` renamed to `ConsumerError` +- `SubscriberApp` renamed to `ConsumerApp` +- `Producer.publish()` method renamed to `Producer.send()` +- `Consumer.subscribe()` method renamed to `Consumer.serve()` + +**@effect-messaging/amqp:** +- `AMQPPublisher` renamed to `AMQPProducer` +- `AMQPSubscriber` renamed to `AMQPConsumer` +- `AMQPSubscriberResponse` renamed to `AMQPConsumerResponse` +- `AMQPProducer.publish()` method renamed to `AMQPProducer.send()` +- `AMQPConsumer.subscribe()` method renamed to `AMQPConsumer.serve()` + +**@effect-messaging/nats:** +- `NATSPublisher` renamed to `NATSProducer` +- `NATSSubscriber` renamed to `NATSConsumer` +- `JetStreamPublisher` renamed to `JetStreamProducer` +- `JetStreamSubscriber` renamed to `JetStreamConsumer` +- `JetStreamSubscriberResponse` renamed to `JetStreamConsumerResponse` +- The previous low-level `JetStreamConsumer` wrapper (for NATS consumers) is now exported as `JetStreamConsumerMessages` +- `NATSProducer.publish()` method renamed to `NATSProducer.send()` +- `JetStreamProducer.publish()` method renamed to `JetStreamProducer.send()` +- `NATSConsumer.subscribe()` method renamed to `NATSConsumer.serve()` +- `JetStreamConsumer.subscribe()` method renamed to `JetStreamConsumer.serve()` + +### Migration Guide + +Update your imports and code references: + +```typescript +// Before +import { Publisher, Subscriber } from "@effect-messaging/core" +import { AMQPPublisher, AMQPSubscriber } from "@effect-messaging/amqp" +import { JetStreamPublisher, JetStreamSubscriber } from "@effect-messaging/nats" + +// After +import { Producer, Consumer } from "@effect-messaging/core" +import { AMQPProducer, AMQPConsumer } from "@effect-messaging/amqp" +import { JetStreamProducer, JetStreamConsumer } from "@effect-messaging/nats" +``` + +Update method calls on producer instances: + +```typescript +// Before +yield* producer.publish({ ... }) + +// After +yield* producer.send({ ... }) +``` + +Update method calls on consumer instances: + +```typescript +// Before +yield* consumer.subscribe(messageHandler) + +// After +yield* consumer.serve(messageHandler) +``` diff --git a/README.md b/README.md index d2042f4..245bcba 100644 --- a/README.md +++ b/README.md @@ -7,13 +7,13 @@ A message broker toolkit for Effect. - 🔌 Effectful wrappers for AMQP Connection and Channel - 🔄 Auto-reconnect functionality when the connection is lost - 🧘 Seamless consumption continuation after reconnection -- 🔭 Distributed tracing support (spans propagate from publishers to subscribers) +- 🔭 Distributed tracing support (spans propagate from producers to consumers) ### NATS / JetStream features - 🔌 Effectful wrappers for NATS Connection and JetStream Client -- 📦 Full JetStream support (streams, consumers, publishers, subscribers) -- 🔭 Distributed tracing support (spans propagate from publishers to subscribers) +- 📦 Full JetStream support (streams, consumers, producers) +- 🔭 Distributed tracing support (spans propagate from producers to consumers) > [!WARNING] > This project is currently **under development**. Please note that future releases might introduce breaking changes. @@ -55,27 +55,27 @@ const runnable = program.pipe( Effect.runPromise(runnable) ``` -#### 2. Create a Publisher +#### 2. Create a Producer -To send messages, create a publisher: +To send messages, create a producer: ```typescript import { AMQPChannel, AMQPConnection, - AMQPPublisher + AMQPProducer } from "@effect-messaging/amqp" import { Context, Effect } from "effect" -class MyPublisher extends Context.Tag("MyPublisher")< - MyPublisher, - AMQPPublisher.AMQPPublisher +class MyProducer extends Context.Tag("MyProducer")< + MyProducer, + AMQPProducer.AMQPProducer >() {} const program = Effect.gen(function* (_) { - const publisher = yield* MyPublisher + const producer = yield* MyProducer - yield* publisher.publish({ + yield* producer.send({ exchange: "my-exchange", routingKey: "my-routing-key", content: Buffer.from('{ "hello": "world" }'), @@ -91,7 +91,7 @@ const program = Effect.gen(function* (_) { }) const runnable = program.pipe( - Effect.provideServiceEffect(MyPublisher, AMQPPublisher.make()), + Effect.provideServiceEffect(MyProducer, AMQPProducer.make()), // provide the AMQP Channel dependency Effect.provide(AMQPChannel.layer), // provide the AMQP Connection dependency @@ -110,19 +110,21 @@ const runnable = program.pipe( Effect.runPromise(runnable) ``` -#### 3. Create a Subscriber +#### 3. Create a Consumer -To receive messages, create a subscriber: +To receive messages, create a consumer. There are two approaches: + +**Option A: Using `serve()` (Layer-based, recommended for production)** ```typescript import { AMQPChannel, AMQPConnection, AMQPConsumeMessage, - AMQPSubscriber, - AMQPSubscriberResponse + AMQPConsumer, + AMQPConsumerResponse } from "@effect-messaging/amqp" -import { Effect } from "effect" +import { Effect, Layer } from "effect" const messageHandler = Effect.gen(function* (_) { const message = yield* AMQPConsumeMessage.AMQPConsumeMessage @@ -134,29 +136,50 @@ const messageHandler = Effect.gen(function* (_) { // - ack(): Acknowledge successful processing // - nack({ allUpTo?, requeue? }): Negative acknowledge // - reject({ requeue? }): Reject the message - return AMQPSubscriberResponse.ack() + return AMQPConsumerResponse.ack() +}) + +// Create a Layer that manages the consumer lifecycle +const ConsumerLive = Layer.unwrapEffect( + Effect.gen(function* (_) { + const consumer = yield* AMQPConsumer.make("my-queue") + return consumer.serve(messageHandler) + }) +) + +const ConnectionLive = AMQPConnection.layer({ + hostname: "localhost", + port: 5672, + username: "guest", + password: "guest", + heartbeat: 10 }) +// Run the consumer as a long-running service +Effect.runPromise( + Layer.launch(ConsumerLive).pipe( + Effect.provide(AMQPChannel.layer), + Effect.provide(ConnectionLive) + ) +) +``` + +**Option B: Using `serveEffect()` (Effect-based, useful for scripts or tests)** + +```typescript const program = Effect.gen(function* (_) { - const subscriber = yield* AMQPSubscriber.make("my-queue") + const consumer = yield* AMQPConsumer.make("my-queue") - // Subscribe to messages - on handler error, messages are nacked automatically - yield* subscriber.subscribe(messageHandler) + // Serve messages - on handler error, messages are nacked automatically + yield* consumer.serveEffect(messageHandler) }) const runnable = program.pipe( + Effect.scoped, // provide the AMQP Channel dependency Effect.provide(AMQPChannel.layer), // provide the AMQP Connection dependency - Effect.provide( - AMQPConnection.layer({ - hostname: "localhost", - port: 5672, - username: "guest", - password: "guest", - heartbeat: 10 - }) - ) + Effect.provide(ConnectionLive) ) // Run the program @@ -186,22 +209,22 @@ const runnable = program.pipe( Effect.runPromise(runnable) ``` -#### 2. Create a JetStream Publisher +#### 2. Create a JetStream Producer To publish messages to a JetStream stream: ```typescript import { JetStreamClient, - JetStreamPublisher, + JetStreamProducer, NATSConnection } from "@effect-messaging/nats" import { Effect } from "effect" const program = Effect.gen(function* (_) { - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() - yield* publisher.publish({ + yield* producer.send({ subject: "orders.created", payload: new TextEncoder().encode('{ "orderId": "123" }') }) @@ -215,19 +238,21 @@ const runnable = program.pipe( Effect.runPromise(runnable) ``` -#### 3. Create a JetStream Subscriber +#### 3. Create a JetStream Consumer + +To consume messages from a JetStream consumer. There are two approaches: -To consume messages from a JetStream consumer: +**Option A: Using `serve()` (Layer-based, recommended for production)** ```typescript import { JetStreamClient, JetStreamMessage, - JetStreamSubscriber, - JetStreamSubscriberResponse, + JetStreamConsumer, + JetStreamConsumerResponse, NATSConnection } from "@effect-messaging/nats" -import { Effect } from "effect" +import { Effect, Layer } from "effect" const messageHandler = Effect.gen(function* (_) { const message = yield* JetStreamMessage.JetStreamConsumeMessage @@ -238,18 +263,46 @@ const messageHandler = Effect.gen(function* (_) { // - ack(): Acknowledge successful processing // - nak({ millis? }): Negative acknowledge, optionally delay redelivery // - term({ reason? }): Terminate message, stop redelivery - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) +// Create a Layer that manages the consumer lifecycle +const ConsumerLive = Layer.unwrapEffect( + Effect.gen(function* (_) { + const client = yield* JetStreamClient.JetStreamClient + + // Get an existing consumer (stream and consumer must already exist) + const natsConsumer = yield* client.consumers.get("my-stream", "my-consumer") + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + + return consumer.serve(messageHandler) + }) +) + +const NATSLive = NATSConnection.layerNode({ servers: ["localhost:4222"] }) +const JetStreamLive = JetStreamClient.layer() + +// Run the consumer as a long-running service +Effect.runPromise( + Layer.launch(ConsumerLive).pipe( + Effect.provide(JetStreamLive), + Effect.provide(NATSLive) + ) +) +``` + +**Option B: Using `serveEffect()` (Effect-based, useful for scripts or tests)** + +```typescript const program = Effect.gen(function* (_) { const client = yield* JetStreamClient.JetStreamClient // Get an existing consumer (stream and consumer must already exist) - const consumer = yield* client.consumers.get("my-stream", "my-consumer") - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) + const natsConsumer = yield* client.consumers.get("my-stream", "my-consumer") + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) - // Subscribe to messages - on handler error, messages are nacked automatically - yield* subscriber.subscribe(messageHandler) + // Serve messages - on handler error, messages are nacked automatically + yield* consumer.serveEffect(messageHandler) }) const runnable = program.pipe( @@ -267,8 +320,8 @@ Effect.runPromise(runnable) **Basic abstractions:** -- [x] Add a `Publisher` interface -- [x] Add a `Subscriber` interface +- [x] Add a `Producer` interface +- [x] Add a `Consumer` interface **Application-level API for consumer apps:** @@ -278,21 +331,21 @@ Effect.runPromise(runnable) **Higher-level declarative API:** - [ ] Add declarative API to define messages schemas -- [ ] Generate publisher based on message definitions +- [ ] Generate producer based on message definitions - [ ] Generate consumer app based on message definitions - [ ] AsyncAPI specification generation ### AMQP implementation - [x] Effect wrappers for AMQP Connection & AMQP Channel -- [x] Implement publisher and subscriber +- [x] Implement producer and consumer - [x] Integration tests - [x] Add examples & documentation ### NATS implementation - [x] Effect wrappers for `@nats-io/nats-core` and `@nats-io/jetstream` -- [x] Implement publisher and subscriber +- [x] Implement producer and consumer - [x] Integration tests - [x] Add examples & documentation diff --git a/packages/amqp/examples/consumer.ts b/packages/amqp/examples/consumer.ts new file mode 100644 index 0000000..80e80f1 --- /dev/null +++ b/packages/amqp/examples/consumer.ts @@ -0,0 +1,62 @@ +import { + AMQPChannel, + AMQPConnection, + AMQPConsumeMessage, + AMQPConsumer, + AMQPConsumerResponse +} from "@effect-messaging/amqp" +import { Effect, Layer } from "effect" + +const messageHandler = Effect.gen(function*(_) { + const message = yield* AMQPConsumeMessage.AMQPConsumeMessage + + // You can add your message processing logic here + yield* Effect.logInfo(`Received message: ${message.content.toString()}`) + + // Return the response to indicate how the message should be handled + return AMQPConsumerResponse.ack() +}) + +// Example 1: Using serve() which returns a Layer (recommended for production) +// The Layer will manage the consumer lifecycle automatically +const ConsumerLive = Layer.unwrapEffect( + Effect.gen(function*(_) { + const consumer = yield* AMQPConsumer.make("my-queue") + return consumer.serve(messageHandler) + }) +) + +const ConnectionLive = AMQPConnection.layer({ + hostname: "localhost", + port: 5672, + username: "guest", + password: "guest", + heartbeat: 10 +}) + +// To run the Layer-based program (recommended for production): +// Effect.runPromise(Layer.launch(ConsumerLive).pipe( +// Effect.provide(AMQPChannel.layer()), +// Effect.provide(ConnectionLive) +// )) + +// Example 2: Using serveEffect() which returns an Effect (useful for scripts or tests) +const effectBasedProgram = Effect.gen(function*(_) { + const consumer = yield* AMQPConsumer.make("my-queue") + + // The consumer will handle message ack/nack/reject based on the response returned by the handler + // On handler failure, the message will be nacked + yield* consumer.serveEffect(messageHandler) +}).pipe( + Effect.scoped, + // provide the AMQP Channel dependency + Effect.provide(AMQPChannel.layer()), + // provide the AMQP Connection dependency + Effect.provide(ConnectionLive) +) + +// Run the Effect-based program +Effect.runPromise(effectBasedProgram) + +// Export ConsumerLive so it can be used elsewhere (e.g., composed with other layers) +export { ConsumerLive } diff --git a/packages/amqp/examples/publisher.ts b/packages/amqp/examples/producer.ts similarity index 70% rename from packages/amqp/examples/publisher.ts rename to packages/amqp/examples/producer.ts index 2d7c4a7..535aacd 100644 --- a/packages/amqp/examples/publisher.ts +++ b/packages/amqp/examples/producer.ts @@ -1,12 +1,12 @@ -import { AMQPChannel, AMQPConnection, AMQPPublisher } from "@effect-messaging/amqp" +import { AMQPChannel, AMQPConnection, AMQPProducer } from "@effect-messaging/amqp" import { Context, Effect } from "effect" -class MyPublisher extends Context.Tag("MyPublisher")() {} +class MyProducer extends Context.Tag("MyProducer")() {} const program = Effect.gen(function*(_) { - const publisher = yield* MyPublisher + const producer = yield* MyProducer - yield* publisher.publish({ + yield* producer.send({ exchange: "my-exchange", routingKey: "my-routing-key", content: Buffer.from("{ \"hello\": \"world\" }"), @@ -22,7 +22,7 @@ const program = Effect.gen(function*(_) { }) const runnable = program.pipe( - Effect.provideServiceEffect(MyPublisher, AMQPPublisher.make()), + Effect.provideServiceEffect(MyProducer, AMQPProducer.make()), // provide the AMQP Channel dependency Effect.provide(AMQPChannel.layer()), // provide the AMQP Connection dependency diff --git a/packages/amqp/examples/subscriber.ts b/packages/amqp/examples/subscriber.ts deleted file mode 100644 index c82578a..0000000 --- a/packages/amqp/examples/subscriber.ts +++ /dev/null @@ -1,42 +0,0 @@ -import { - AMQPChannel, - AMQPConnection, - AMQPConsumeMessage, - AMQPSubscriber, - AMQPSubscriberResponse -} from "@effect-messaging/amqp" -import { Effect } from "effect" - -const messageHandler = Effect.gen(function*(_) { - const message = yield* AMQPConsumeMessage.AMQPConsumeMessage - - // You can add your message processing logic here - yield* Effect.logInfo(`Received message: ${message.content.toString()}`) - - // Return the response to indicate how the message should be handled - return AMQPSubscriberResponse.ack() -}) - -const program = Effect.gen(function*(_) { - const subscriber = yield* AMQPSubscriber.make("my-queue") - - // The subscriber will handle message ack/nack/reject based on the response returned by the handler - // On handler failure, the message will be nacked - yield* subscriber.subscribe(messageHandler) -}) - -const runnable = program.pipe( - // provide the AMQP Channel dependency - Effect.provide(AMQPChannel.layer()), - // provide the AMQP Connection dependency - Effect.provide(AMQPConnection.layer({ - hostname: "localhost", - port: 5672, - username: "guest", - password: "guest", - heartbeat: 10 - })) -) - -// Run the program -Effect.runPromise(runnable) diff --git a/packages/amqp/src/AMQPConsumer.ts b/packages/amqp/src/AMQPConsumer.ts new file mode 100644 index 0000000..4d3af27 --- /dev/null +++ b/packages/amqp/src/AMQPConsumer.ts @@ -0,0 +1,251 @@ +/** + * @since 0.3.0 + */ +import * as Consumer from "@effect-messaging/core/Consumer" +import type * as ConsumerApp from "@effect-messaging/core/ConsumerApp" +import * as ConsumerError from "@effect-messaging/core/ConsumerError" +import * as Headers from "@effect/platform/Headers" +import * as HttpTraceContext from "@effect/platform/HttpTraceContext" +import type { Options } from "amqplib" +import * as Cause from "effect/Cause" +import type * as Duration from "effect/Duration" +import * as Effect from "effect/Effect" +import * as Function from "effect/Function" +import * as Layer from "effect/Layer" +import * as Match from "effect/Match" +import * as Option from "effect/Option" +import * as Predicate from "effect/Predicate" +import type * as Scope from "effect/Scope" +import * as Stream from "effect/Stream" +import * as AMQPChannel from "./AMQPChannel.js" +import type * as AMQPConnection from "./AMQPConnection.js" +import * as AMQPConsumeMessage from "./AMQPConsumeMessage.js" +import type * as AMQPConsumerResponse from "./AMQPConsumerResponse.js" +import type * as AMQPError from "./AMQPError.js" + +/** + * @category type ids + * @since 0.3.0 + */ +export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPConsumer") + +/** + * @category type ids + * @since 0.3.0 + */ +export type TypeId = typeof TypeId + +/** + * @category models + * @since 0.3.0 + */ +export interface AMQPPublishMessage { + exchange: string + routingKey: string + content: Buffer + options?: Options.Publish +} + +/** + * @category models + * @since 0.5.0 + */ +export type AMQPConsumerApp = ConsumerApp.ConsumerApp< + AMQPConsumerResponse.AMQPConsumerResponse, + AMQPConsumeMessage.AMQPConsumeMessage, + E, + R +> + +/** + * @category models + * @since 0.3.0 + */ +export interface AMQPConsumer + extends Consumer.Consumer +{ + readonly [TypeId]: TypeId +} + +const ATTR_SERVER_ADDRESS = "server.address" as const +const ATTR_SERVER_PORT = "server.port" as const +const ATTR_MESSAGING_DESTINATION_NAME = "messaging.destination.name" as const +const ATTR_MESSAGING_OPERATION_NAME = "messaging.operation.name" as const +const ATTR_MESSAGING_OPERATION_TYPE = "messaging.operation.type" as const +const ATTR_MESSAGING_SYSTEM = "messaging.system" as const +const ATTR_MESSAGING_MESSAGE_ID = "messaging.message.id" as const +const ATTR_MESSAGING_MESSAGE_CONVERSATION_ID = "messaging.message.conversation_id" as const +const ATTR_MESSAGING_AMQP_DESTINATION_ROUTING_KEY = "messaging.amqp.destination.routing_key" as const +const ATTR_MESSAGING_AMQP_MESSAGE_DELIVERY_TAG = "messaging.amqp.message.delivery_tag" as const +const ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME = "messaging.destination.subscription.name" as const + +/** @internal */ +const serveEffect = ( + channel: AMQPChannel.AMQPChannel, + queueName: string, + connectionProperties: AMQPConnection.AMQPConnectionServerProperties, + options: AMQPConsumerOptions +) => +( + app: AMQPConsumerApp +): Effect.Effect< + void, + ConsumerError.ConsumerError, + Scope.Scope | Exclude> +> => { + const handleMessage = (message: AMQPConsumeMessage.AMQPConsumeMessage) => + Effect.fork( + Effect.useSpan( + `amqp.consume ${message.fields.routingKey}`, + { + parent: Option.getOrUndefined( + HttpTraceContext.fromHeaders(Headers.fromInput(message.properties.headers)) + ), + kind: "consumer", + captureStackTrace: false, + attributes: { + [ATTR_SERVER_ADDRESS]: connectionProperties.host, + [ATTR_SERVER_PORT]: connectionProperties.port, + [ATTR_MESSAGING_MESSAGE_ID]: message.properties.messageId, + [ATTR_MESSAGING_MESSAGE_CONVERSATION_ID]: message.properties.correlationId, + [ATTR_MESSAGING_SYSTEM]: connectionProperties.product, + [ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME]: queueName, + [ATTR_MESSAGING_DESTINATION_NAME]: queueName, + [ATTR_MESSAGING_OPERATION_TYPE]: "receive", + [ATTR_MESSAGING_AMQP_DESTINATION_ROUTING_KEY]: message.fields.routingKey, + [ATTR_MESSAGING_AMQP_MESSAGE_DELIVERY_TAG]: message.fields.deliveryTag + } + }, + (span) => + Effect.gen(function*() { + yield* Effect.logDebug(`amqp.consume ${message.fields.routingKey}`) + return yield* app.pipe( + options.handlerTimeout + ? Effect.timeoutFail({ + duration: options.handlerTimeout, + onTimeout: () => new ConsumerError.ConsumerError({ reason: `AMQPConsumer: handler timed out` }) + }) + : Function.identity + ) + }).pipe( + Effect.provide(AMQPConsumeMessage.layer(message)), + Effect.matchCauseEffect({ + onSuccess: (response) => + Match.valueTags(response, { + Ack: () => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "ack") + yield* channel.ack(message) + }), + Nack: (r) => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nack") + yield* channel.nack(message, r.allUpTo, r.requeue) + }), + Reject: (r) => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "reject") + yield* channel.reject(message, r.requeue) + }) + }), + onFailure: (cause) => + Effect.gen(function*() { + yield* Effect.logError(Cause.pretty(cause)) + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nack") + span.attribute( + "error.type", + Cause.squashWith( + cause, + (_) => Predicate.hasProperty(_, "tag") ? _.tag : _ instanceof Error ? _.name : `${_}` + ) + ) + span.attribute("error.stack", Cause.pretty(cause)) + span.attribute( + "error.message", + Cause.squashWith( + cause, + (_) => Predicate.hasProperty(_, "reason") ? _.reason : _ instanceof Error ? _.message : `${_}` + ) + ) + yield* channel.nack(message, false, false) + }) + }), + options.uninterruptible ? Effect.uninterruptible : Effect.interruptible, + Effect.withParentSpan(span) + ) + ) + ) + + return channel.consume( + queueName, + options.concurrency ? { prefetch: options.concurrency } : undefined + ).pipe( + Effect.flatMap((consumeStream) => Stream.runForEach(consumeStream, handleMessage)), + Effect.mapError((error) => + new ConsumerError.ConsumerError({ reason: `AMQPConsumer failed to subscribe`, cause: error }) + ) + ) +} + +/** @internal */ +const serveLayer = ( + channel: AMQPChannel.AMQPChannel, + queueName: string, + connectionProperties: AMQPConnection.AMQPConnectionServerProperties, + options: AMQPConsumerOptions +) => +( + app: AMQPConsumerApp +): Layer.Layer< + never, + ConsumerError.ConsumerError, + Exclude> +> => Layer.scopedDiscard(serveEffect(channel, queueName, connectionProperties, options)(app)) + +/** @internal */ +const healthCheck = ( + channel: AMQPChannel.AMQPChannel, + queueName: string +): Effect.Effect => + channel.checkQueue(queueName).pipe( + Effect.catchTag("AMQPChannelError", (error) => + new ConsumerError.ConsumerError({ reason: `Healthcheck failed`, cause: error })), + Effect.asVoid + ) + +/** + * @category models + * @since 0.5.0 + */ +export interface AMQPConsumerOptions { + uninterruptible?: boolean + handlerTimeout?: Duration.DurationInput + concurrency?: number +} + +/** + * @category constructors + * @since 0.3.0 + */ +export const make = ( + queueName: string, + options: AMQPConsumerOptions = {} +): Effect.Effect< + AMQPConsumer, + AMQPError.AMQPConnectionError, + AMQPChannel.AMQPChannel +> => + Effect.gen(function*() { + const channel = yield* AMQPChannel.AMQPChannel + const serverProperties = yield* channel.connection.serverProperties + + const consumer: AMQPConsumer = { + [TypeId]: TypeId, + [Consumer.TypeId]: Consumer.TypeId, + serve: serveLayer(channel, queueName, serverProperties, options), + serveEffect: serveEffect(channel, queueName, serverProperties, options), + healthCheck: healthCheck(channel, queueName) + } + + return consumer + }) diff --git a/packages/amqp/src/AMQPSubscriberResponse.ts b/packages/amqp/src/AMQPConsumerResponse.ts similarity index 78% rename from packages/amqp/src/AMQPSubscriberResponse.ts rename to packages/amqp/src/AMQPConsumerResponse.ts index e1226fe..fc16b89 100644 --- a/packages/amqp/src/AMQPSubscriberResponse.ts +++ b/packages/amqp/src/AMQPConsumerResponse.ts @@ -6,7 +6,7 @@ * @category type ids * @since 0.5.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPSubscriberResponse") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPConsumerResponse") /** * @category type ids @@ -65,7 +65,7 @@ export interface Reject { * @category models * @since 0.5.0 */ -export type AMQPSubscriberResponse = Ack | Nack | Reject +export type AMQPConsumerResponse = Ack | Nack | Reject class AckImpl implements Ack { readonly [TypeId]: TypeId = TypeId @@ -93,23 +93,23 @@ class RejectImpl implements Reject { * @category constructors * @since 0.5.0 */ -export const ack = (): AMQPSubscriberResponse => new AckImpl() +export const ack = (): AMQPConsumerResponse => new AckImpl() /** * @category constructors * @since 0.5.0 */ -export const nack = (options?: NackOptions): AMQPSubscriberResponse => new NackImpl(options?.allUpTo, options?.requeue) +export const nack = (options?: NackOptions): AMQPConsumerResponse => new NackImpl(options?.allUpTo, options?.requeue) /** * @category constructors * @since 0.5.0 */ -export const reject = (options?: RejectOptions): AMQPSubscriberResponse => new RejectImpl(options?.requeue) +export const reject = (options?: RejectOptions): AMQPConsumerResponse => new RejectImpl(options?.requeue) /** * @category guards * @since 0.5.0 */ -export const isAMQPSubscriberResponse = (u: unknown): u is AMQPSubscriberResponse => +export const isAMQPConsumerResponse = (u: unknown): u is AMQPConsumerResponse => typeof u === "object" && u !== null && TypeId in u diff --git a/packages/amqp/src/AMQPPublisher.ts b/packages/amqp/src/AMQPProducer.ts similarity index 63% rename from packages/amqp/src/AMQPPublisher.ts rename to packages/amqp/src/AMQPProducer.ts index d491822..38a139b 100644 --- a/packages/amqp/src/AMQPPublisher.ts +++ b/packages/amqp/src/AMQPProducer.ts @@ -1,8 +1,8 @@ /** * @since 0.3.0 */ -import * as Publisher from "@effect-messaging/core/Publisher" -import * as PublisherError from "@effect-messaging/core/PublisherError" +import * as Producer from "@effect-messaging/core/Producer" +import * as ProducerError from "@effect-messaging/core/ProducerError" import type { Options } from "amqplib" import * as Effect from "effect/Effect" import * as Schedule from "effect/Schedule" @@ -13,7 +13,7 @@ import type * as AMQPError from "./AMQPError.js" * @category type ids * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPPublisher") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPProducer") /** * @category type ids @@ -36,7 +36,7 @@ export interface AMQPPublishMessage { * @category models * @since 0.3.0 */ -export interface AMQPPublisher extends Publisher.Publisher { +export interface AMQPProducer extends Producer.Producer { readonly [TypeId]: TypeId } @@ -45,12 +45,12 @@ const publish = ( channel: AMQPChannel.AMQPChannel, retrySchedule: Schedule.Schedule ) => -(message: AMQPPublishMessage): Effect.Effect => +(message: AMQPPublishMessage): Effect.Effect => channel.publish(message.exchange, message.routingKey, message.content, message.options).pipe( Effect.retry(retrySchedule), Effect.catchTag( "AMQPChannelError", - (error) => Effect.fail(new PublisherError.PublisherError({ reason: "Failed to publish message", cause: error })) + (error) => Effect.fail(new ProducerError.ProducerError({ reason: "Failed to publish message", cause: error })) ), Effect.map(() => undefined) ) @@ -59,7 +59,7 @@ const publish = ( * @category constructors * @since 0.3.2 */ -export interface AMQPPublisherConfig { +export interface AMQPProducerConfig { readonly retrySchedule?: Schedule.Schedule } @@ -67,15 +67,15 @@ export interface AMQPPublisherConfig { * @category constructors * @since 0.3.0 */ -export const make = (config?: AMQPPublisherConfig): Effect.Effect => +export const make = (config?: AMQPProducerConfig): Effect.Effect => Effect.gen(function*() { const channel = yield* AMQPChannel.AMQPChannel - const publisher: AMQPPublisher = { + const producer: AMQPProducer = { [TypeId]: TypeId, - [Publisher.TypeId]: Publisher.TypeId, - publish: publish(channel, config?.retrySchedule ?? Schedule.stop) + [Producer.TypeId]: Producer.TypeId, + send: publish(channel, config?.retrySchedule ?? Schedule.stop) } - return publisher + return producer }) diff --git a/packages/amqp/src/AMQPSubscriber.ts b/packages/amqp/src/AMQPSubscriber.ts deleted file mode 100644 index 79f55a1..0000000 --- a/packages/amqp/src/AMQPSubscriber.ts +++ /dev/null @@ -1,227 +0,0 @@ -/** - * @since 0.3.0 - */ -import * as Subscriber from "@effect-messaging/core/Subscriber" -import type * as SubscriberApp from "@effect-messaging/core/SubscriberApp" -import * as SubscriberError from "@effect-messaging/core/SubscriberError" -import * as Headers from "@effect/platform/Headers" -import * as HttpTraceContext from "@effect/platform/HttpTraceContext" -import type { Options } from "amqplib" -import * as Cause from "effect/Cause" -import type * as Duration from "effect/Duration" -import * as Effect from "effect/Effect" -import * as Function from "effect/Function" -import * as Match from "effect/Match" -import * as Option from "effect/Option" -import * as Predicate from "effect/Predicate" -import * as Stream from "effect/Stream" -import * as AMQPChannel from "./AMQPChannel.js" -import type * as AMQPConnection from "./AMQPConnection.js" -import * as AMQPConsumeMessage from "./AMQPConsumeMessage.js" -import type * as AMQPError from "./AMQPError.js" -import type * as AMQPSubscriberResponse from "./AMQPSubscriberResponse.js" - -/** - * @category type ids - * @since 0.3.0 - */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/amqp/AMQPSubscriber") - -/** - * @category type ids - * @since 0.3.0 - */ -export type TypeId = typeof TypeId - -/** - * @category models - * @since 0.3.0 - */ -export interface AMQPPublishMessage { - exchange: string - routingKey: string - content: Buffer - options?: Options.Publish -} - -/** - * @category models - * @since 0.5.0 - */ -export type AMQPSubscriberApp = SubscriberApp.SubscriberApp< - AMQPSubscriberResponse.AMQPSubscriberResponse, - AMQPConsumeMessage.AMQPConsumeMessage, - E, - R -> - -/** - * @category models - * @since 0.3.0 - */ -export interface AMQPSubscriber - extends Subscriber.Subscriber -{ - readonly [TypeId]: TypeId -} - -const ATTR_SERVER_ADDRESS = "server.address" as const -const ATTR_SERVER_PORT = "server.port" as const -const ATTR_MESSAGING_DESTINATION_NAME = "messaging.destination.name" as const -const ATTR_MESSAGING_OPERATION_NAME = "messaging.operation.name" as const -const ATTR_MESSAGING_OPERATION_TYPE = "messaging.operation.type" as const -const ATTR_MESSAGING_SYSTEM = "messaging.system" as const -const ATTR_MESSAGING_MESSAGE_ID = "messaging.message.id" as const -const ATTR_MESSAGING_MESSAGE_CONVERSATION_ID = "messaging.message.conversation_id" as const -const ATTR_MESSAGING_AMQP_DESTINATION_ROUTING_KEY = "messaging.amqp.destination.routing_key" as const -const ATTR_MESSAGING_AMQP_MESSAGE_DELIVERY_TAG = "messaging.amqp.message.delivery_tag" as const -const ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME = "messaging.destination.subscription.name" as const - -/** @internal */ -const subscribe = ( - channel: AMQPChannel.AMQPChannel, - queueName: string, - connectionProperties: AMQPConnection.AMQPConnectionServerProperties, - options: AMQPSubscriberOptions -) => -(app: AMQPSubscriberApp) => - Effect.gen(function*() { - const consumeStream = yield* channel.consume( - queueName, - options.concurrency ? { prefetch: options.concurrency } : undefined - ) - return yield* consumeStream.pipe( - Stream.runForEach((message) => - Effect.fork( - Effect.useSpan( - `amqp.consume ${message.fields.routingKey}`, - { - parent: Option.getOrUndefined( - HttpTraceContext.fromHeaders(Headers.fromInput(message.properties.headers)) - ), - kind: "consumer", - captureStackTrace: false, - attributes: { - [ATTR_SERVER_ADDRESS]: connectionProperties.host, - [ATTR_SERVER_PORT]: connectionProperties.port, - [ATTR_MESSAGING_MESSAGE_ID]: message.properties.messageId, - [ATTR_MESSAGING_MESSAGE_CONVERSATION_ID]: message.properties.correlationId, - [ATTR_MESSAGING_SYSTEM]: connectionProperties.product, - [ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME]: queueName, - [ATTR_MESSAGING_DESTINATION_NAME]: queueName, - [ATTR_MESSAGING_OPERATION_TYPE]: "receive", - [ATTR_MESSAGING_AMQP_DESTINATION_ROUTING_KEY]: message.fields.routingKey, - [ATTR_MESSAGING_AMQP_MESSAGE_DELIVERY_TAG]: message.fields.deliveryTag - } - }, - (span) => - Effect.gen(function*() { - yield* Effect.logDebug(`amqp.consume ${message.fields.routingKey}`) - const response = yield* app.pipe( - options.handlerTimeout - ? Effect.timeoutFail({ - duration: options.handlerTimeout, - onTimeout: () => - new SubscriberError.SubscriberError({ reason: `AMQPSubscriber: handler timed out` }) - }) - : Function.identity - ) - yield* Match.valueTags(response, { - Ack: () => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "ack") - yield* channel.ack(message) - }), - Nack: (r) => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nack") - yield* channel.nack(message, r.allUpTo, r.requeue) - }), - Reject: (r) => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "reject") - yield* channel.reject(message, r.requeue) - }) - }) - }).pipe( - Effect.provide(AMQPConsumeMessage.layer(message)), - Effect.tapErrorCause((cause) => - Effect.gen(function*() { - yield* Effect.logError(Cause.pretty(cause)) - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nack") - span.attribute( - "error.type", - String(Cause.squashWith( - cause, - (_) => Predicate.hasProperty(_, "_tag") ? _._tag : _ instanceof Error ? _.name : `${_}` - )) - ) - span.attribute("error.stack", Cause.pretty(cause)) - span.attribute( - "error.message", - String(Cause.squashWith( - cause, - (_) => Predicate.hasProperty(_, "reason") ? _.reason : _ instanceof Error ? _.message : `${_}` - )) - ) - yield* channel.nack(message, false, false) - }) - ), - options.uninterruptible ? Effect.uninterruptible : Effect.interruptible, - Effect.withParentSpan(span) - ) - ) - ) - ), - Effect.mapError((error) => - new SubscriberError.SubscriberError({ reason: `AMQPSubscriber failed to subscribe`, cause: error }) - ) - ) - }) - -/** @internal */ -const healthCheck = ( - channel: AMQPChannel.AMQPChannel, - queueName: string -): Effect.Effect => - channel.checkQueue(queueName).pipe( - Effect.catchTag("AMQPChannelError", (error) => - new SubscriberError.SubscriberError({ reason: `Healthcheck failed`, cause: error })), - Effect.asVoid - ) - -/** - * @category models - * @since 0.5.0 - */ -export interface AMQPSubscriberOptions { - uninterruptible?: boolean - handlerTimeout?: Duration.DurationInput - concurrency?: number -} - -/** - * @category constructors - * @since 0.3.0 - */ -export const make = ( - queueName: string, - options: AMQPSubscriberOptions = {} -): Effect.Effect< - AMQPSubscriber, - AMQPError.AMQPConnectionError, - AMQPChannel.AMQPChannel -> => - Effect.gen(function*() { - const channel = yield* AMQPChannel.AMQPChannel - const serverProperties = yield* channel.connection.serverProperties - - const subscriber: AMQPSubscriber = { - [TypeId]: TypeId, - [Subscriber.TypeId]: Subscriber.TypeId, - subscribe: subscribe(channel, queueName, serverProperties, options), - healthCheck: healthCheck(channel, queueName) - } - - return subscriber - }) diff --git a/packages/amqp/src/index.ts b/packages/amqp/src/index.ts index c40f074..20b3b74 100644 --- a/packages/amqp/src/index.ts +++ b/packages/amqp/src/index.ts @@ -14,21 +14,21 @@ export * as AMQPConnection from "./AMQPConnection.js" export * as AMQPConsumeMessage from "./AMQPConsumeMessage.js" /** - * @since 0.1.0 + * @since 0.3.0 */ -export * as AMQPError from "./AMQPError.js" +export * as AMQPConsumer from "./AMQPConsumer.js" /** - * @since 0.3.0 + * @since 0.5.0 */ -export * as AMQPPublisher from "./AMQPPublisher.js" +export * as AMQPConsumerResponse from "./AMQPConsumerResponse.js" /** - * @since 0.3.0 + * @since 0.1.0 */ -export * as AMQPSubscriber from "./AMQPSubscriber.js" +export * as AMQPError from "./AMQPError.js" /** - * @since 0.5.0 + * @since 0.3.0 */ -export * as AMQPSubscriberResponse from "./AMQPSubscriberResponse.js" +export * as AMQPProducer from "./AMQPProducer.js" diff --git a/packages/amqp/test/AMQPSubscriber.test.ts b/packages/amqp/test/AMQPConsumer.test.ts similarity index 52% rename from packages/amqp/test/AMQPSubscriber.test.ts rename to packages/amqp/test/AMQPConsumer.test.ts index e42f702..04b8e43 100644 --- a/packages/amqp/test/AMQPSubscriber.test.ts +++ b/packages/amqp/test/AMQPConsumer.test.ts @@ -1,11 +1,11 @@ import type { Mock } from "@effect/vitest" import { describe, expect, it, vi } from "@effect/vitest" -import { Effect, Schedule, TestServices } from "effect" +import { Effect, Layer, Schedule, TestServices } from "effect" import * as AMQPChannel from "../src/AMQPChannel.js" import * as AMQPConsumeMessage from "../src/AMQPConsumeMessage.js" -import * as AMQPPublisher from "../src/AMQPPublisher.js" -import * as AMQPSubscriber from "../src/AMQPSubscriber.js" -import * as AMQPSubscriberResponse from "../src/AMQPSubscriberResponse.js" +import * as AMQPConsumer from "../src/AMQPConsumer.js" +import * as AMQPConsumerResponse from "../src/AMQPConsumerResponse.js" +import * as AMQPProducer from "../src/AMQPProducer.js" import { assertTestExchange, assertTestQueue, @@ -20,15 +20,15 @@ import { } from "./dependencies.js" const publishAndAssertConsume = ( - { content, onMessage, publisher, times }: { - publisher: AMQPPublisher.AMQPPublisher + { content, onMessage, producer, times }: { + producer: AMQPProducer.AMQPProducer onMessage: Mock<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void> content: Buffer times: number } ) => Effect.gen(function*() { - yield* publisher.publish({ + yield* producer.send({ exchange: TEST_EXCHANGE, routingKey: TEST_SUBJECT, content @@ -53,32 +53,32 @@ const setup = Effect.gen(function*() { yield* purgeTestQueue }) -describe("AMQPChannel", { sequential: true }, () => { - describe("subscribe", () => { +describe("AMQPConsumer", { sequential: true }, () => { + describe("serve", () => { it.effect("Should consume published events even when connection or channel fails", () => - Effect.gen(function*() { + Effect.scoped(Effect.gen(function*() { yield* setup - const publisher = yield* AMQPPublisher.make({ + const producer = yield* AMQPProducer.make({ retrySchedule: Schedule.exponential("100 millis", 1.5).pipe( Schedule.jittered, Schedule.intersect(Schedule.recurs(10)) ) }) - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE) + const consumer = yield* AMQPConsumer.make(TEST_QUEUE) const onMessage = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() - // Start the subscription - yield* Effect.fork(subscriber.subscribe(Effect.gen(function*() { + // Start the subscription using Layer.launch + yield* Effect.fork(Layer.launch(consumer.serve(Effect.gen(function*() { const message = yield* AMQPConsumeMessage.AMQPConsumeMessage onMessage(message) - return AMQPSubscriberResponse.ack() - }))) + return AMQPConsumerResponse.ack() + })))) // Message 1 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 1"), times: 1 @@ -86,7 +86,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 2 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 2"), times: 2 @@ -97,7 +97,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 3 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 3"), times: 3 @@ -105,7 +105,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 4 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 4"), times: 4 @@ -116,7 +116,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 5 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 5"), times: 5 @@ -124,7 +124,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 6 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 6"), times: 6 @@ -135,7 +135,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 7 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 7"), times: 7 @@ -143,22 +143,22 @@ describe("AMQPChannel", { sequential: true }, () => { // Message 8 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: Buffer.from("Message 8"), times: 8 }) - }).pipe(Effect.provide(testChannel), TestServices.provideLive)) + })).pipe(Effect.provide(testChannel), TestServices.provideLive)) }) - describe("interruptable subscribers", { sequential: true }, () => { + describe("interruptable consumers", { sequential: true }, () => { it.effect( "Should interrupt the handler if the subscription fiber is interrupted, and the message should be consumed again", () => - Effect.gen(function*() { + Effect.scoped(Effect.gen(function*() { yield* setup - const publisher = yield* AMQPPublisher.make() + const producer = yield* AMQPProducer.make() const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() const onHandlingFinished = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() @@ -168,18 +168,18 @@ describe("AMQPChannel", { sequential: true }, () => { onHandlingStarted(message) yield* Effect.sleep("500 millis") onHandlingFinished(message) - return AMQPSubscriberResponse.ack() + return AMQPConsumerResponse.ack() }) - const startSubscription = Effect.gen(function*() { - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE) - yield* subscriber.subscribe(handler) - }).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription + const startSubscription = Effect.scoped(Effect.gen(function*() { + const consumer = yield* AMQPConsumer.make(TEST_QUEUE) + return yield* Layer.launch(consumer.serve(handler)) + })).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription // Start the subscription - const subscribptionFiber1 = yield* Effect.fork(startSubscription) + const subscriptionFiber1 = yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ exchange: TEST_EXCHANGE, routingKey: TEST_SUBJECT, content: Buffer.from("My Message that will be interrupted") @@ -190,7 +190,7 @@ describe("AMQPChannel", { sequential: true }, () => { // Verify the message was consumed expect(onHandlingStarted).toHaveBeenCalledTimes(1) - yield* subscribptionFiber1.interruptAsFork(subscribptionFiber1.id()) + yield* subscriptionFiber1.interruptAsFork(subscriptionFiber1.id()) // Wait for the interruption to complete yield* Effect.sleep("500 millis") @@ -205,69 +205,73 @@ describe("AMQPChannel", { sequential: true }, () => { // The same message should be consumed again because the first subscription was interrupted and the message was nor acked nor nacked expect(onHandlingStarted).toHaveBeenCalledTimes(2) expect(onHandlingFinished).toHaveBeenCalledTimes(1) - }).pipe(Effect.provide(testChannel), TestServices.provideLive), + })).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 15000 } ) - it.effect("Should no interrupt the handler if the subscriber is uninterruptible", () => - Effect.gen(function*() { - yield* setup + it.effect( + "Should no interrupt the handler if the consumer is uninterruptible", + () => + Effect.scoped(Effect.gen(function*() { + yield* setup - const publisher = yield* AMQPPublisher.make() + const producer = yield* AMQPProducer.make() - const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() - const onHandlingFinished = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() + const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() + const onHandlingFinished = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() - const handler = Effect.gen(function*() { - const message = yield* AMQPConsumeMessage.AMQPConsumeMessage - onHandlingStarted(message) - yield* Effect.sleep("300 millis") - onHandlingFinished(message) - return AMQPSubscriberResponse.ack() - }) + const handler = Effect.gen(function*() { + const message = yield* AMQPConsumeMessage.AMQPConsumeMessage + onHandlingStarted(message) + yield* Effect.sleep("300 millis") + onHandlingFinished(message) + return AMQPConsumerResponse.ack() + }) - const startSubscription = Effect.gen(function*() { - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE, { uninterruptible: true }) - yield* subscriber.subscribe(handler) - }).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription + const startSubscription = Effect.scoped(Effect.gen(function*() { + const consumer = yield* AMQPConsumer.make(TEST_QUEUE, { uninterruptible: true }) + return yield* Layer.launch(consumer.serve(handler)) + })).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription - // Start the subscription - const subscribptionFiber1 = yield* Effect.fork(startSubscription) + // Start the subscription + const subscriptionFiber1 = yield* Effect.fork(startSubscription) - yield* publisher.publish({ - exchange: TEST_EXCHANGE, - routingKey: TEST_SUBJECT, - content: Buffer.from("My Message that will NOT be interrupted") - }) + yield* producer.send({ + exchange: TEST_EXCHANGE, + routingKey: TEST_SUBJECT, + content: Buffer.from("My Message that will NOT be interrupted") + }) - // Wait for the message to be consumed - yield* Effect.sleep("200 millis") - // Verify the message was consumed - expect(onHandlingStarted).toHaveBeenCalledTimes(1) + // Wait for the message to be consumed + yield* Effect.sleep("200 millis") + // Verify the message was consumed + expect(onHandlingStarted).toHaveBeenCalledTimes(1) - // Interrupt the subscription fiber - yield* subscribptionFiber1.interruptAsFork(subscribptionFiber1.id()) + // Interrupt the subscription fiber + yield* subscriptionFiber1.interruptAsFork(subscriptionFiber1.id()) - // The subscription should be uninterrupted - wait for the message to be consumed - yield* Effect.sleep("300 millis") - expect(onHandlingFinished).toHaveBeenCalledTimes(1) + // The subscription should be uninterrupted - wait for the message to be consumed + yield* Effect.sleep("300 millis") + expect(onHandlingFinished).toHaveBeenCalledTimes(1) - // Start the subscription again (with a new channel) - yield* Effect.fork(startSubscription) + // Start the subscription again (with a new channel) + yield* Effect.fork(startSubscription) - yield* Effect.sleep("500 millis") - // The same message should not be consumed again because the first subscription was uninterrupted and the message was acked or nacked - expect(onHandlingStarted).toHaveBeenCalledTimes(1) - expect(onHandlingFinished).toHaveBeenCalledTimes(1) - }).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 15000 }) + yield* Effect.sleep("500 millis") + // The same message should not be consumed again because the first subscription was uninterrupted and the message was acked or nacked + expect(onHandlingStarted).toHaveBeenCalledTimes(1) + expect(onHandlingFinished).toHaveBeenCalledTimes(1) + })).pipe(Effect.provide(testChannel), TestServices.provideLive), + { timeout: 15000 } + ) it.effect( - "Should interrupt the handler if the subscriber is uninterruptible but reaches the timeout", + "Should interrupt the handler if the consumer is uninterruptible but reaches the timeout", () => - Effect.gen(function*() { + Effect.scoped(Effect.gen(function*() { yield* setup - const publisher = yield* AMQPPublisher.make() + const producer = yield* AMQPProducer.make() const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() const onHandlingFinished = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() @@ -278,21 +282,21 @@ describe("AMQPChannel", { sequential: true }, () => { // long running task yield* Effect.sleep("500 millis") onHandlingFinished(message) - return AMQPSubscriberResponse.ack() + return AMQPConsumerResponse.ack() }) - const startSubscription = Effect.gen(function*() { - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE, { + const startSubscription = Effect.scoped(Effect.gen(function*() { + const consumer = yield* AMQPConsumer.make(TEST_QUEUE, { uninterruptible: true, handlerTimeout: "300 millis" }) - yield* subscriber.subscribe(handler) - }).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription + return yield* Layer.launch(consumer.serve(handler)) + })).pipe(Effect.provide(AMQPChannel.layer())) // Provide a fresh channel for each subscription // Start the subscription - const subscribptionFiber1 = yield* Effect.fork(startSubscription) + const subscriptionFiber1 = yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ exchange: TEST_EXCHANGE, routingKey: TEST_SUBJECT, content: Buffer.from("My Message that will NOT be interrupted") @@ -304,7 +308,7 @@ describe("AMQPChannel", { sequential: true }, () => { expect(onHandlingStarted).toHaveBeenCalledTimes(1) // Interrupt the subscription fiber - yield* subscribptionFiber1.interruptAsFork(subscribptionFiber1.id()) + yield* subscriptionFiber1.interruptAsFork(subscriptionFiber1.id()) // wait for the handler to timeout yield* Effect.sleep("300 millis") @@ -318,17 +322,17 @@ describe("AMQPChannel", { sequential: true }, () => { // The same message should not be consumed again because the has timed out and was nacked expect(onHandlingStarted).toHaveBeenCalledTimes(1) expect(onHandlingFinished).toHaveBeenCalledTimes(1) - }).pipe(Effect.provide(testChannel), TestServices.provideLive), + })).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 30000 } ) }) describe("explicit response types", () => { it.effect("Should nack the message when handler returns nack()", () => - Effect.gen(function*() { + Effect.scoped(Effect.gen(function*() { yield* setup - const publisher = yield* AMQPPublisher.make() + const producer = yield* AMQPProducer.make() const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() let attemptCount = 0 @@ -340,22 +344,22 @@ describe("AMQPChannel", { sequential: true }, () => { if (attemptCount === 1) { // First attempt: nack with requeue to trigger redelivery - return AMQPSubscriberResponse.nack({ requeue: true }) + return AMQPConsumerResponse.nack({ requeue: true }) } // Subsequent attempts: ack - return AMQPSubscriberResponse.ack() + return AMQPConsumerResponse.ack() }) - const startSubscription = Effect.gen(function*() { - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE) - yield* subscriber.subscribe(handler) - }).pipe(Effect.provide(AMQPChannel.layer())) + const startSubscription = Effect.scoped(Effect.gen(function*() { + const consumer = yield* AMQPConsumer.make(TEST_QUEUE) + return yield* Layer.launch(consumer.serve(handler)) + })).pipe(Effect.provide(AMQPChannel.layer())) // Start the subscription yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ exchange: TEST_EXCHANGE, routingKey: TEST_SUBJECT, content: Buffer.from("Message that will be nacked first") @@ -366,47 +370,51 @@ describe("AMQPChannel", { sequential: true }, () => { // The message should be processed twice: once nacked, once acked expect(onHandlingStarted).toHaveBeenCalledTimes(2) - }).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 15000 }) + })).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 15000 }) - it.effect("Should reject the message without requeue when handler returns reject()", () => - Effect.gen(function*() { - yield* setup + it.effect( + "Should reject the message without requeue when handler returns reject()", + () => + Effect.scoped(Effect.gen(function*() { + yield* setup - const publisher = yield* AMQPPublisher.make() + const producer = yield* AMQPProducer.make() - const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() + const onHandlingStarted = vi.fn<(message: AMQPConsumeMessage.AMQPConsumeMessage) => void>() - const handler = Effect.gen(function*() { - const message = yield* AMQPConsumeMessage.AMQPConsumeMessage - onHandlingStarted(message) + const handler = Effect.gen(function*() { + const message = yield* AMQPConsumeMessage.AMQPConsumeMessage + onHandlingStarted(message) - // Reject the message without requeue - message won't be redelivered - return AMQPSubscriberResponse.reject({ requeue: false }) - }) + // Reject the message without requeue - message won't be redelivered + return AMQPConsumerResponse.reject({ requeue: false }) + }) - const startSubscription = Effect.gen(function*() { - const subscriber = yield* AMQPSubscriber.make(TEST_QUEUE) - yield* subscriber.subscribe(handler) - }).pipe(Effect.provide(AMQPChannel.layer())) + const startSubscription = Effect.scoped(Effect.gen(function*() { + const consumer = yield* AMQPConsumer.make(TEST_QUEUE) + return yield* Layer.launch(consumer.serve(handler)) + })).pipe(Effect.provide(AMQPChannel.layer())) - // Start the subscription - yield* Effect.fork(startSubscription) + // Start the subscription + yield* Effect.fork(startSubscription) - yield* publisher.publish({ - exchange: TEST_EXCHANGE, - routingKey: TEST_SUBJECT, - content: Buffer.from("Message that will be rejected") - }) + yield* producer.send({ + exchange: TEST_EXCHANGE, + routingKey: TEST_SUBJECT, + content: Buffer.from("Message that will be rejected") + }) - // Wait for message to be processed - yield* Effect.sleep("500 millis") + // Wait for message to be processed + yield* Effect.sleep("500 millis") - // The message should be processed only once (reject without requeue prevents redelivery) - expect(onHandlingStarted).toHaveBeenCalledTimes(1) + // The message should be processed only once (reject without requeue prevents redelivery) + expect(onHandlingStarted).toHaveBeenCalledTimes(1) - // Wait a bit more to ensure no redelivery happens - yield* Effect.sleep("1 second") - expect(onHandlingStarted).toHaveBeenCalledTimes(1) - }).pipe(Effect.provide(testChannel), TestServices.provideLive), { timeout: 15000 }) + // Wait a bit more to ensure no redelivery happens + yield* Effect.sleep("1 second") + expect(onHandlingStarted).toHaveBeenCalledTimes(1) + })).pipe(Effect.provide(testChannel), TestServices.provideLive), + { timeout: 15000 } + ) }) }) diff --git a/packages/core/src/Consumer.ts b/packages/core/src/Consumer.ts new file mode 100644 index 0000000..5d933f9 --- /dev/null +++ b/packages/core/src/Consumer.ts @@ -0,0 +1,64 @@ +/** + * @since 0.3.0 + */ +import type * as Effect from "effect/Effect" +import type * as Layer from "effect/Layer" +import type * as Scope from "effect/Scope" +import type * as ConsumerApp from "./ConsumerApp.js" +import type * as ConsumerError from "./ConsumerError.js" + +/** + * @category type ids + * @since 0.3.0 + */ +export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Consumer") + +/** + * @category type ids + * @since 0.3.0 + */ +export type TypeId = typeof TypeId + +/** + * Types that are automatically provided to consumer handlers. + * + * @typeParam M - The message type provided via Effect context + * + * @category models + * @since 0.8.0 + */ +export type Provided = M + +/** + * @category models + * @since 0.3.0 + */ +export interface Consumer { + readonly [TypeId]: TypeId + + /** + * Serve the consumer application as a Layer. + * + * The returned Layer will run the consumer until the layer is released. + * This is the recommended way to run consumers in production. + * + * @since 0.8.0 + */ + readonly serve: ( + app: ConsumerApp.ConsumerApp + ) => Layer.Layer>> + + /** + * Serve the consumer application as an Effect. + * + * The returned Effect will run the consumer until interrupted or an error occurs. + * Requires a Scope to manage the consumer lifecycle. + * + * @since 0.8.0 + */ + readonly serveEffect: ( + app: ConsumerApp.ConsumerApp + ) => Effect.Effect>> + + readonly healthCheck: Effect.Effect +} diff --git a/packages/core/src/SubscriberApp.ts b/packages/core/src/ConsumerApp.ts similarity index 74% rename from packages/core/src/SubscriberApp.ts rename to packages/core/src/ConsumerApp.ts index 7667764..806f789 100644 --- a/packages/core/src/SubscriberApp.ts +++ b/packages/core/src/ConsumerApp.ts @@ -4,7 +4,7 @@ import type * as Effect from "effect/Effect" /** - * Type alias for a subscriber handler function that processes messages. + * Type alias for a consumer handler function that processes messages. * * @typeParam A - The response type that the handler must return (e.g., acknowledgment response) * @typeParam M - The message type that the handler receives via Effect context @@ -14,4 +14,4 @@ import type * as Effect from "effect/Effect" * @since 0.3.0 * @category models */ -export type SubscriberApp = Effect.Effect +export type ConsumerApp = Effect.Effect diff --git a/packages/core/src/PublisherError.ts b/packages/core/src/ConsumerError.ts similarity index 71% rename from packages/core/src/PublisherError.ts rename to packages/core/src/ConsumerError.ts index 095b1a9..18f2dd1 100644 --- a/packages/core/src/PublisherError.ts +++ b/packages/core/src/ConsumerError.ts @@ -6,7 +6,7 @@ import * as Schema from "effect/Schema" /** * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/PublisherError") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/ConsumerError") /** * @since 0.3.0 @@ -14,13 +14,13 @@ export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Publishe export type TypeId = typeof TypeId /** - * Represents a generic Publisher Error + * Represents a generic Consumer Error * * @since 0.3.0 * @category errors */ -export class PublisherError extends Schema.TaggedError()( - "PublisherError", +export class ConsumerError extends Schema.TaggedError()( + "ConsumerError", { reason: Schema.String, cause: Schema.optional(Schema.Defect) } ) { /** diff --git a/packages/core/src/ConsumerMiddleware.ts b/packages/core/src/ConsumerMiddleware.ts new file mode 100644 index 0000000..c26cdb1 --- /dev/null +++ b/packages/core/src/ConsumerMiddleware.ts @@ -0,0 +1,36 @@ +/** + * Middleware abstraction for consumer applications. + * + * Middleware allows you to transform consumer applications by wrapping + * them with additional logic such as logging, error handling, or metrics. + * + * @since 0.8.0 + */ +import type * as ConsumerApp from "./ConsumerApp.js" + +/** + * A middleware transforms a consumer application into another consumer application. + * + * @typeParam A - The response type that handlers must return + * @typeParam M - The message type provided via Effect context + * @typeParam EIn - The error type of the input application + * @typeParam RIn - The dependencies of the input application + * @typeParam EOut - The error type of the output application (defaults to EIn) + * @typeParam ROut - The dependencies of the output application (defaults to RIn) + * + * @category models + * @since 0.8.0 + */ +export type ConsumerMiddleware = ( + app: ConsumerApp.ConsumerApp +) => ConsumerApp.ConsumerApp + +/** + * Create a middleware from a function that transforms a consumer application. + * + * @category constructors + * @since 0.8.0 + */ +export const make = ( + f: (app: ConsumerApp.ConsumerApp) => ConsumerApp.ConsumerApp +): ConsumerMiddleware => f diff --git a/packages/core/src/Publisher.ts b/packages/core/src/Producer.ts similarity index 62% rename from packages/core/src/Publisher.ts rename to packages/core/src/Producer.ts index 70d0006..5194f8f 100644 --- a/packages/core/src/Publisher.ts +++ b/packages/core/src/Producer.ts @@ -2,13 +2,13 @@ * @since 0.3.0 */ import type * as Effect from "effect/Effect" -import type * as PublisherError from "./PublisherError.js" +import type * as ProducerError from "./ProducerError.js" /** * @category type ids * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Publisher") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Producer") /** * @category type ids @@ -20,7 +20,7 @@ export type TypeId = typeof TypeId * @category models * @since 0.3.0 */ -export interface Publisher { +export interface Producer { readonly [TypeId]: TypeId - readonly publish: (message: M) => Effect.Effect + readonly send: (message: M) => Effect.Effect } diff --git a/packages/core/src/SubscriberError.ts b/packages/core/src/ProducerError.ts similarity index 70% rename from packages/core/src/SubscriberError.ts rename to packages/core/src/ProducerError.ts index babf786..114dc63 100644 --- a/packages/core/src/SubscriberError.ts +++ b/packages/core/src/ProducerError.ts @@ -6,7 +6,7 @@ import * as Schema from "effect/Schema" /** * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/SubscriberError") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/ProducerError") /** * @since 0.3.0 @@ -14,13 +14,13 @@ export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Subscrib export type TypeId = typeof TypeId /** - * Represents a generic Publisher Error + * Represents a generic Producer Error * * @since 0.3.0 * @category errors */ -export class SubscriberError extends Schema.TaggedError()( - "SubscriberError", +export class ProducerError extends Schema.TaggedError()( + "ProducerError", { reason: Schema.String, cause: Schema.optional(Schema.Defect) } ) { /** diff --git a/packages/core/src/Subscriber.ts b/packages/core/src/Subscriber.ts deleted file mode 100644 index 78d00ae..0000000 --- a/packages/core/src/Subscriber.ts +++ /dev/null @@ -1,30 +0,0 @@ -/** - * @since 0.3.0 - */ -import type * as Effect from "effect/Effect" -import type * as SubscriberApp from "./SubscriberApp.js" -import type * as SubscriberError from "./SubscriberError.js" - -/** - * @category type ids - * @since 0.3.0 - */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/core/Subscriber") - -/** - * @category type ids - * @since 0.3.0 - */ -export type TypeId = typeof TypeId - -/** - * @category models - * @since 0.3.0 - */ -export interface Subscriber { - readonly [TypeId]: TypeId - readonly subscribe: ( - app: SubscriberApp.SubscriberApp - ) => Effect.Effect> - readonly healthCheck: Effect.Effect -} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d692c02..dca6e37 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -1,24 +1,34 @@ /** * @since 0.3.0 */ -export * as Publisher from "./Publisher.js" +export * as Consumer from "./Consumer.js" /** * @since 0.3.0 */ -export * as PublisherError from "./PublisherError.js" +export * as ConsumerApp from "./ConsumerApp.js" /** * @since 0.3.0 */ -export * as Subscriber from "./Subscriber.js" +export * as ConsumerError from "./ConsumerError.js" + +/** + * Middleware abstraction for consumer applications. + * + * Middleware allows you to transform consumer applications by wrapping + * them with additional logic such as logging, error handling, or metrics. + * + * @since 0.8.0 + */ +export * as ConsumerMiddleware from "./ConsumerMiddleware.js" /** * @since 0.3.0 */ -export * as SubscriberApp from "./SubscriberApp.js" +export * as Producer from "./Producer.js" /** * @since 0.3.0 */ -export * as SubscriberError from "./SubscriberError.js" +export * as ProducerError from "./ProducerError.js" diff --git a/packages/core/test/Producer.test.ts b/packages/core/test/Producer.test.ts new file mode 100644 index 0000000..f2ee382 --- /dev/null +++ b/packages/core/test/Producer.test.ts @@ -0,0 +1,8 @@ +import { assert, describe, it } from "@effect/vitest" +import * as Producer from "../src/Producer.js" + +describe("Producer", () => { + it("Dummy test", () => { + assert.equal(Producer.TypeId, Producer.TypeId) + }) +}) diff --git a/packages/core/test/Publisher.test.ts b/packages/core/test/Publisher.test.ts deleted file mode 100644 index b8785ad..0000000 --- a/packages/core/test/Publisher.test.ts +++ /dev/null @@ -1,8 +0,0 @@ -import { assert, describe, it } from "@effect/vitest" -import * as Publisher from "../src/Publisher.js" - -describe("Publisher", () => { - it("Dummy test", () => { - assert.equal(Publisher.TypeId, Publisher.TypeId) - }) -}) diff --git a/packages/nats/AGENTS.md b/packages/nats/AGENTS.md index d7baf8c..143a138 100644 --- a/packages/nats/AGENTS.md +++ b/packages/nats/AGENTS.md @@ -7,7 +7,13 @@ Effect bindings for NATS and JetStream. This package mimics the architecture of **Key modules:** - `NATSConnection` - Core NATS connection (publish/subscribe/request) -- `JetStreamClient` - JetStream consumer and publisher +- `JetStreamClient` - JetStream client for stream and consumer access +- `JetStreamProducer` - Producer implementation for JetStream (implements `@effect-messaging/core/Producer`) +- `JetStreamConsumer` - Consumer implementation for JetStream (implements `@effect-messaging/core/Consumer`) +- `JetStreamConsumerMessages` - Low-level NATS consumer wrapper with Effect operations +- `JetStreamConsumerResponse` - Response types for consumer message handling (ack/nak/term) +- `NATSProducer` - Producer implementation for NATS Core (implements `@effect-messaging/core/Producer`) +- `NATSConsumer` - Consumer implementation for NATS Core (implements `@effect-messaging/core/Consumer`) - `JetStreamConsumerAPI` - Consumer management API - `JetStreamStreamAPI` - Stream management API - `JetStreamDirectStreamAPI` - Direct stream API for low-latency reads @@ -29,6 +35,28 @@ Effect bindings for NATS and JetStream. This package mimics the architecture of - Avoid `any` types - use proper type casting with internal types when needed - Never use `null` - always use `Option` from Effect for optional values +## Consumer API Pattern + +Consumers (`JetStreamConsumer`, `NATSConsumer`) follow the `@effect-messaging/core/Consumer` interface with two methods for serving handlers: + +- `serve(handler)` - Returns a `Layer`. Recommended for production as it integrates with the Effect Layer system for lifecycle management. +- `serveEffect(handler)` - Returns an `Effect`. Useful for scripts, tests, or when you need direct Effect control. + +Example usage: + +```typescript +// Layer-based (production) +const ConsumerLive = Layer.unwrapEffect( + Effect.gen(function* () { + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + return consumer.serve(messageHandler) + }) +) + +// Effect-based (scripts/tests) +yield * consumer.serveEffect(messageHandler) +``` + ## Testing - Tests require a running NATS server at `localhost:4222` (use `docker-compose up -d` from root) diff --git a/packages/nats/examples/consumer.ts b/packages/nats/examples/consumer.ts new file mode 100644 index 0000000..f3873c1 --- /dev/null +++ b/packages/nats/examples/consumer.ts @@ -0,0 +1,83 @@ +import { + JetStreamClient, + JetStreamConsumer, + JetStreamConsumerResponse, + JetStreamMessage, + NATSConnection +} from "@effect-messaging/nats" +import { Effect, Layer } from "effect" + +const messageHandler = Effect.gen(function*() { + const message = yield* JetStreamMessage.JetStreamConsumeMessage + + // Parse and process the message + const data = message.string() + yield* Effect.logInfo(`Received message on ${message.subject}: ${data}`) + + // Return a response to acknowledge, nak, or terminate the message + // - ack(): Acknowledge successful processing + // - nak({ millis? }): Negative acknowledge, optionally delay redelivery + // - term({ reason? }): Terminate message, stop redelivery + return JetStreamConsumerResponse.ack() +}) + +// Example 1: Using serve() which returns a Layer (recommended for production) +// The Layer will manage the consumer lifecycle automatically +const ConsumerLive = Layer.unwrapEffect( + Effect.gen(function*() { + const client = yield* JetStreamClient.JetStreamClient + + // Get an existing consumer from a stream + // Note: The stream and consumer must already exist in NATS + const natsConsumer = yield* client.consumers.get("my-stream", "my-consumer") + + // Create a consumer from the NATS consumer + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer, { + // Optional: make message processing uninterruptible + uninterruptible: true, + // Optional: set a timeout for message processing + handlerTimeout: "30 seconds" + }) + + return consumer.serve(messageHandler) + }) +) + +// Create layers for the NATS connection and JetStream client +const NATSConnectionLive = NATSConnection.layerNode({ servers: ["localhost:4222"] }) +const JetStreamClientLive = JetStreamClient.layer() + +// To run the Layer-based program (recommended for production): +// Effect.runPromise(Layer.launch(ConsumerLive).pipe( +// Effect.provide(JetStreamClientLive), +// Effect.provide(NATSConnectionLive) +// )) + +// Example 2: Using serveEffect() which returns an Effect (useful for scripts or tests) +const effectBasedProgram = Effect.gen(function*() { + const client = yield* JetStreamClient.JetStreamClient + + // Get an existing consumer from a stream + const natsConsumer = yield* client.consumers.get("my-stream", "my-consumer") + + // Create a consumer from the NATS consumer + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer, { + uninterruptible: true, + handlerTimeout: "30 seconds" + }) + + yield* Effect.logInfo("Starting to consume messages...") + + // Subscribe to messages - this will run until interrupted + yield* consumer.serveEffect(messageHandler) +}).pipe( + Effect.scoped, + Effect.provide(JetStreamClientLive), + Effect.provide(NATSConnectionLive) +) + +// Run the Effect-based program +Effect.runPromise(effectBasedProgram) + +// Export ConsumerLive so it can be used elsewhere (e.g., composed with other layers) +export { ConsumerLive } diff --git a/packages/nats/examples/publisher.ts b/packages/nats/examples/producer.ts similarity index 63% rename from packages/nats/examples/publisher.ts rename to packages/nats/examples/producer.ts index 23acdd1..aa0687a 100644 --- a/packages/nats/examples/publisher.ts +++ b/packages/nats/examples/producer.ts @@ -1,12 +1,12 @@ -import { JetStreamClient, JetStreamPublisher, NATSConnection } from "@effect-messaging/nats" +import { JetStreamClient, JetStreamProducer, NATSConnection } from "@effect-messaging/nats" import { Context, Effect, Layer } from "effect" -class MyPublisher extends Context.Tag("MyPublisher")() {} +class MyProducer extends Context.Tag("MyProducer")() {} const program = Effect.gen(function*() { - const publisher = yield* MyPublisher + const producer = yield* MyProducer - yield* publisher.publish({ + yield* producer.send({ subject: "orders.created", payload: JSON.stringify({ orderId: "12345", amount: 99.99 }), options: { @@ -17,15 +17,15 @@ const program = Effect.gen(function*() { yield* Effect.logInfo("Message published successfully") }) -// Create a layer that provides the publisher -const PublisherLive = Layer.effect(MyPublisher, JetStreamPublisher.make()) +// Create a layer that provides the producer +const ProducerLive = Layer.effect(MyProducer, JetStreamProducer.make()) // Create layers for the NATS connection and JetStream client const NATSConnectionLive = NATSConnection.layerNode({ servers: ["localhost:4222"] }) const JetStreamClientLive = JetStreamClient.layer() // Compose all layers -const MainLive = PublisherLive.pipe( +const MainLive = ProducerLive.pipe( Layer.provide(JetStreamClientLive), Layer.provide(NATSConnectionLive) ) diff --git a/packages/nats/examples/subscriber.ts b/packages/nats/examples/subscriber.ts deleted file mode 100644 index 3e8d6fb..0000000 --- a/packages/nats/examples/subscriber.ts +++ /dev/null @@ -1,55 +0,0 @@ -import { - JetStreamClient, - JetStreamMessage, - JetStreamSubscriber, - JetStreamSubscriberResponse, - NATSConnection -} from "@effect-messaging/nats" -import { Effect } from "effect" - -const messageHandler = Effect.gen(function*() { - const message = yield* JetStreamMessage.JetStreamConsumeMessage - - // Parse and process the message - const data = message.string() - yield* Effect.logInfo(`Received message on ${message.subject}: ${data}`) - - // Return a response to acknowledge, nak, or terminate the message - // - ack(): Acknowledge successful processing - // - nak({ millis? }): Negative acknowledge, optionally delay redelivery - // - term({ reason? }): Terminate message, stop redelivery - return JetStreamSubscriberResponse.ack() -}) - -const program = Effect.gen(function*() { - const client = yield* JetStreamClient.JetStreamClient - - // Get an existing consumer from a stream - // Note: The stream and consumer must already exist in NATS - const consumer = yield* client.consumers.get("my-stream", "my-consumer") - - // Create a subscriber from the consumer - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer, { - // Optional: make message processing uninterruptible - uninterruptible: true, - // Optional: set a timeout for message processing - handlerTimeout: "30 seconds" - }) - - yield* Effect.logInfo("Starting to consume messages...") - - // Subscribe to messages - this will run until interrupted - yield* subscriber.subscribe(messageHandler) -}) - -// Create layers for the NATS connection and JetStream client -const NATSConnectionLive = NATSConnection.layerNode({ servers: ["localhost:4222"] }) -const JetStreamClientLive = JetStreamClient.layer() - -const runnable = program.pipe( - Effect.provide(JetStreamClientLive), - Effect.provide(NATSConnectionLive) -) - -// Run the program -Effect.runPromise(runnable) diff --git a/packages/nats/src/JetStreamClient.ts b/packages/nats/src/JetStreamClient.ts index e43efe6..b2c5271 100644 --- a/packages/nats/src/JetStreamClient.ts +++ b/packages/nats/src/JetStreamClient.ts @@ -7,7 +7,7 @@ import * as Effect from "effect/Effect" import * as Layer from "effect/Layer" import * as utils from "./internal/utils.js" import * as JetStreamBatch from "./JetStreamBatch.js" -import * as JetStreamConsumers from "./JetStreamConsumer.js" +import * as JetStreamConsumerMessages from "./JetStreamConsumerMessages.js" import * as JetStreamStream from "./JetStreamStream.js" import * as NATSConnection from "./NATSConnection.js" import * as NATSError from "./NATSError.js" @@ -40,7 +40,7 @@ export interface JetStreamClient { ...params: Parameters ) => Effect.Effect readonly options: Effect.Effect - readonly consumers: JetStreamConsumers.Consumers + readonly consumers: JetStreamConsumerMessages.Consumers readonly streams: JetStreamStream.JetStreamStreams /** @internal */ @@ -67,7 +67,7 @@ export const make = (js: JetStream.JetStreamClient): JetStreamClient => ({ Effect.map(JetStreamBatch.make) ), options: wrap(() => js.getOptions(), "Failed to get JetStream options"), - consumers: JetStreamConsumers.makeConsumers(js.consumers), + consumers: JetStreamConsumerMessages.makeConsumers(js.consumers), streams: JetStreamStream.makeJetStreamStreams(js.streams), js }) diff --git a/packages/nats/src/JetStreamConsumer.ts b/packages/nats/src/JetStreamConsumer.ts index 112e497..9d19bb2 100644 --- a/packages/nats/src/JetStreamConsumer.ts +++ b/packages/nats/src/JetStreamConsumer.ts @@ -1,366 +1,261 @@ /** * @since 0.1.0 */ -import type * as JetStream from "@nats-io/jetstream" +import * as Consumer from "@effect-messaging/core/Consumer" +import type * as ConsumerApp from "@effect-messaging/core/ConsumerApp" +import * as ConsumerError from "@effect-messaging/core/ConsumerError" +import type * as NATSCore from "@nats-io/nats-core" +import * as Cause from "effect/Cause" +import type * as Duration from "effect/Duration" import * as Effect from "effect/Effect" +import * as Function from "effect/Function" +import * as Layer from "effect/Layer" +import * as Match from "effect/Match" import * as Option from "effect/Option" +import * as Predicate from "effect/Predicate" +import type * as Scope from "effect/Scope" import * as Stream from "effect/Stream" -import * as utils from "./internal/utils.js" +import type * as JetStreamConsumerMessages from "./JetStreamConsumerMessages.js" +import type * as JetStreamConsumerResponse from "./JetStreamConsumerResponse.js" import * as JetStreamMessage from "./JetStreamMessage.js" +import * as NATSConnection from "./NATSConnection.js" import * as NATSError from "./NATSError.js" -import * as NATSQueuedIterator from "./NATSQueuedIterator.js" - -const wrap = utils.wrap(NATSError.JetStreamConsumerError) -const wrapAsync = utils.wrapAsync(NATSError.JetStreamConsumerError) +import * as NATSHeaders from "./NATSHeaders.js" /** * @category type ids * @since 0.1.0 */ -export const CloseTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Close") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamConsumer") /** * @category type ids * @since 0.1.0 */ -export type CloseTypeId = typeof CloseTypeId +export type TypeId = typeof TypeId /** - * Represents closeable resources - * * @category models - * @since 0.1.0 - */ -export interface Close { - readonly [CloseTypeId]: CloseTypeId - readonly close: Effect.Effect - readonly closed: Effect.Effect -} - -/** @internal */ -export const makeClose = (closeable: JetStream.Close): Close => ({ - [CloseTypeId]: CloseTypeId, - close: wrapAsync(() => closeable.close(), "Failed to close resource").pipe( - Effect.flatMap((result) => - result instanceof Error - ? Effect.fail(new NATSError.JetStreamConsumerError({ reason: "Failed to close resource", cause: result })) - : Effect.void - ) - ), - closed: wrapAsync(() => closeable.closed(), "Failed to check closed status").pipe( - Effect.flatMap((result) => - result instanceof Error - ? Effect.fail(new NATSError.JetStreamConsumerError({ reason: "Resource closed with error", cause: result })) - : Effect.void - ) - ) -}) - -/** - * @category type ids - * @since 0.1.0 - */ -export const ConsumerMessagesTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ConsumerMessages") - -/** - * @category type ids - * @since 0.1.0 + * @since 0.7.0 */ -export type ConsumerMessagesTypeId = typeof ConsumerMessagesTypeId +export type JetStreamConsumerApp = ConsumerApp.ConsumerApp< + JetStreamConsumerResponse.JetStreamConsumerResponse, + JetStreamMessage.JetStreamMessage, + E, + R +> /** - * Represents consumer messages - * * @category models * @since 0.1.0 */ -export interface ConsumerMessages extends - Omit< - NATSQueuedIterator.NATSQueuedIterator, - "iterator" - >, - Close +export interface JetStreamConsumer + extends Consumer.Consumer { - readonly [ConsumerMessagesTypeId]: ConsumerMessagesTypeId - readonly status: Effect.Effect< - Stream.Stream, - NATSError.JetStreamConsumerError - > - - /** @internal */ - readonly consumerMessages: JetStream.ConsumerMessages -} - -/** @internal */ -export const makeConsumerMessages = (consumerMessages: JetStream.ConsumerMessages): ConsumerMessages => { - const qi = NATSQueuedIterator.make(NATSError.JetStreamConsumerError)(consumerMessages) - const close = makeClose(consumerMessages) - return { - ...qi, - ...close, - [ConsumerMessagesTypeId]: ConsumerMessagesTypeId, - stream: Stream.map(qi.stream, JetStreamMessage.make), - status: wrap( - () => - Stream.fromAsyncIterable( - consumerMessages.status(), - (error) => - new NATSError.JetStreamConsumerError({ - reason: "An error occurred in consumer status async iterable", - cause: error - }) - ), - "Failed to get status stream" - ), - consumerMessages - } -} - -/** - * @category type ids - * @since 0.1.0 - */ -export const ConsumerKindTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ConsumerKind") - -/** - * @category type ids - * @since 0.1.0 - */ -export type ConsumerKindTypeId = typeof ConsumerKindTypeId - -/** - * Represents consumer kind - * - * @category models - * @since 0.1.0 - */ -export interface ConsumerKind { - readonly [ConsumerKindTypeId]: ConsumerKindTypeId - readonly isPullConsumer: Effect.Effect - readonly isPushConsumer: Effect.Effect -} - -/** - * @category type ids - * @since 0.1.0 - */ -export const ExportedConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ExportedConsumer") - -/** - * @category type ids - * @since 0.1.0 - */ -export type ExportedConsumerTypeId = typeof ExportedConsumerTypeId - -/** - * Represents an exported consumer - * - * @category models - * @since 0.1.0 - */ -export interface ExportedConsumer extends ConsumerKind { - readonly [ExportedConsumerTypeId]: ExportedConsumerTypeId - readonly next: ( - ...args: Parameters - ) => Effect.Effect, NATSError.JetStreamConsumerError> - readonly fetch: ( - ...args: Parameters - ) => Effect.Effect - readonly consume: ( - ...args: Parameters - ) => Effect.Effect -} - -/** - * @category type ids - * @since 0.1.0 - */ -export const InfoableConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/InfoableConsumer") - -/** - * @category type ids - * @since 0.1.0 - */ -export type InfoableConsumerTypeId = typeof InfoableConsumerTypeId - -/** - * Represents an infoable consumer - * - * @category models - * @since 0.1.0 - */ -export interface InfoableConsumer { - readonly [InfoableConsumerTypeId]: InfoableConsumerTypeId - readonly info: ( - ...args: Parameters - ) => Effect.Effect + readonly [TypeId]: TypeId } /** - * @category type ids - * @since 0.1.0 - */ -export const DeleteableConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/DeleteableConsumer") - -/** - * @category type ids - * @since 0.1.0 - */ -export type DeleteableConsumerTypeId = typeof DeleteableConsumerTypeId - -/** - * Represents a deleteable consumer - * * @category models * @since 0.1.0 */ -export interface DeleteableConsumer { - readonly [DeleteableConsumerTypeId]: DeleteableConsumerTypeId - readonly delete: Effect.Effect +export interface JetStreamConsumerOptions { + uninterruptible?: boolean + handlerTimeout?: Duration.DurationInput } -/** - * @category type ids - * @since 0.1.0 - */ -export const ConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Consumer") - -/** - * @category type ids - * @since 0.1.0 - */ -export type ConsumerTypeId = typeof ConsumerTypeId - -/** - * Represents a consumer - * - * @category models - * @since 0.1.0 - */ -export interface Consumer extends ExportedConsumer, InfoableConsumer, DeleteableConsumer { - readonly [ConsumerTypeId]: ConsumerTypeId - - /** @internal */ - readonly consumer: JetStream.Consumer -} +const ATTR_SERVER_ADDRESS = "server.address" as const +const ATTR_SERVER_PORT = "server.port" as const +const ATTR_MESSAGING_DESTINATION_NAME = "messaging.destination.name" as const +const ATTR_MESSAGING_OPERATION_NAME = "messaging.operation.name" as const +const ATTR_MESSAGING_OPERATION_TYPE = "messaging.operation.type" as const +const ATTR_MESSAGING_SYSTEM = "messaging.system" as const +const ATTR_MESSAGING_MESSAGE_ID = "messaging.message.id" as const +const ATTR_MESSAGING_NATS_STREAM = "messaging.nats.stream" as const +const ATTR_MESSAGING_NATS_CONSUMER = "messaging.nats.consumer" as const +const ATTR_MESSAGING_NATS_SEQUENCE_STREAM = "messaging.nats.sequence.stream" as const +const ATTR_MESSAGING_NATS_SEQUENCE_CONSUMER = "messaging.nats.sequence.consumer" as const /** @internal */ -export const makeConsumer = (consumer: JetStream.Consumer): Consumer => ({ - [ConsumerTypeId]: ConsumerTypeId, - [ExportedConsumerTypeId]: ExportedConsumerTypeId, - [ConsumerKindTypeId]: ConsumerKindTypeId, - [InfoableConsumerTypeId]: InfoableConsumerTypeId, - [DeleteableConsumerTypeId]: DeleteableConsumerTypeId, - isPullConsumer: wrap(() => consumer.isPullConsumer(), "Failed to check if consumer is pull consumer"), - isPushConsumer: wrap(() => consumer.isPushConsumer(), "Failed to check if consumer is push consumer"), - next: (...args) => - wrapAsync(() => consumer.next(...args), "Failed to get next message").pipe( - Effect.map((msg) => Option.fromNullable(msg ? JetStreamMessage.make(msg) : null)) +const serveEffect = ( + consumerMessages: JetStreamConsumerMessages.ConsumerMessages, + connectionInfo: NATSCore.ServerInfo, + options: JetStreamConsumerOptions +) => +( + app: JetStreamConsumerApp +): Effect.Effect< + void, + ConsumerError.ConsumerError, + Scope.Scope | Exclude> +> => + consumerMessages.stream.pipe( + Stream.runForEach((message) => + Effect.fork( + Effect.useSpan( + `nats.consume ${message.subject}`, + { + parent: Option.getOrUndefined(NATSHeaders.decodeTraceContextOptional(message.headers)), + kind: "consumer", + captureStackTrace: false, + attributes: { + [ATTR_SERVER_ADDRESS]: connectionInfo.host, + [ATTR_SERVER_PORT]: connectionInfo.port, + [ATTR_MESSAGING_SYSTEM]: "nats", + [ATTR_MESSAGING_OPERATION_TYPE]: "receive", + [ATTR_MESSAGING_DESTINATION_NAME]: message.subject, + [ATTR_MESSAGING_MESSAGE_ID]: message.seq, + [ATTR_MESSAGING_NATS_STREAM]: message.info.stream, + [ATTR_MESSAGING_NATS_CONSUMER]: message.info.consumer, + [ATTR_MESSAGING_NATS_SEQUENCE_STREAM]: message.info.streamSequence, + [ATTR_MESSAGING_NATS_SEQUENCE_CONSUMER]: message.info.deliverySequence + } + }, + (span) => + Effect.gen(function*() { + yield* Effect.logDebug(`nats.consume ${message.subject}`) + return yield* app.pipe( + options.handlerTimeout + ? Effect.timeoutFail({ + duration: options.handlerTimeout, + onTimeout: () => new ConsumerError.ConsumerError({ reason: "JetStreamConsumer: handler timed out" }) + }) + : Function.identity + ) + }).pipe( + Effect.provide(JetStreamMessage.layer(message)), + Effect.matchCauseEffect( + { + onSuccess: (res) => + Match.valueTags(res, { + Ack: () => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "ack") + yield* message.ack + }), + Nak: (r) => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nak") + yield* message.nak(r.millis) + }), + Term: (r) => + Effect.gen(function*() { + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "term") + yield* message.term(r.reason) + }) + }), + onFailure: (cause) => + Effect.gen(function*() { + yield* Effect.logError(Cause.pretty(cause)) + span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nak") + span.attribute( + "error.type", + Cause.squashWith( + cause, + (_) => Predicate.hasProperty(_, "tag") ? _.tag : _ instanceof Error ? _.name : `${_}` + ) + ) + span.attribute("error.stack", Cause.pretty(cause)) + span.attribute( + "error.message", + Cause.squashWith( + cause, + (_) => Predicate.hasProperty(_, "reason") ? _.reason : _ instanceof Error ? _.message : `${_}` + ) + ) + yield* message.nak() + }) + } + ), + options.uninterruptible ? Effect.uninterruptible : Effect.interruptible, + Effect.withParentSpan(span) + ) + ) + ) ), - fetch: (...args) => - wrapAsync(() => consumer.fetch(...args), "Failed to fetch messages").pipe(Effect.map(makeConsumerMessages)), - consume: (...args) => - wrapAsync(() => consumer.consume(...args), "Failed to consume messages").pipe(Effect.map(makeConsumerMessages)), - info: (...args) => wrapAsync(() => consumer.info(...args), "Failed to get consumer info"), - delete: wrapAsync(() => consumer.delete(), "Failed to delete consumer"), - consumer -}) - -/** - * @category type ids - * @since 0.1.0 - */ -export const PushConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/PushConsumer") - -/** - * @category type ids - * @since 0.1.0 - */ -export type PushConsumerTypeId = typeof PushConsumerTypeId - -/** - * Represents a push consumer - * - * @category models - * @since 0.1.0 - */ -export interface PushConsumer extends InfoableConsumer, DeleteableConsumer, ConsumerKind { - readonly [PushConsumerTypeId]: PushConsumerTypeId - readonly consume: ( - ...args: Parameters - ) => Effect.Effect - - /** @internal */ - readonly pushConsumer: JetStream.PushConsumer -} + Effect.mapError((error) => + new ConsumerError.ConsumerError({ reason: "JetStreamConsumer failed to subscribe", cause: error }) + ) + ) /** @internal */ -export const makePushConsumer = (pushConsumer: JetStream.PushConsumer): PushConsumer => ({ - [PushConsumerTypeId]: PushConsumerTypeId, - [ConsumerKindTypeId]: ConsumerKindTypeId, - [InfoableConsumerTypeId]: InfoableConsumerTypeId, - [DeleteableConsumerTypeId]: DeleteableConsumerTypeId, - isPullConsumer: wrap(() => pushConsumer.isPullConsumer(), "Failed to check if consumer is pull consumer"), - isPushConsumer: wrap(() => pushConsumer.isPushConsumer(), "Failed to check if consumer is push consumer"), - consume: (...args) => - wrapAsync(() => pushConsumer.consume(...args), "Failed to consume messages").pipe(Effect.map(makeConsumerMessages)), - info: (...args) => wrapAsync(() => pushConsumer.info(...args), "Failed to get consumer info"), - delete: wrapAsync(() => pushConsumer.delete(), "Failed to delete consumer"), - pushConsumer -}) +const serveLayer = ( + consumerMessages: JetStreamConsumerMessages.ConsumerMessages, + connectionInfo: NATSCore.ServerInfo, + options: JetStreamConsumerOptions +) => +( + app: JetStreamConsumerApp +): Layer.Layer< + never, + ConsumerError.ConsumerError, + Exclude> +> => Layer.scopedDiscard(serveEffect(consumerMessages, connectionInfo, options)(app)) -/** - * @category type ids - * @since 0.1.0 - */ -export const ConsumersTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Consumers") +/** @internal */ +const healthCheck = ( + consumer: JetStreamConsumerMessages.InfoableConsumer +): Effect.Effect => + consumer.info().pipe( + Effect.catchTag("JetStreamConsumerError", (error) => + new ConsumerError.ConsumerError({ reason: "Healthcheck failed", cause: error })), + Effect.asVoid + ) /** - * @category type ids + * Create a JetStreamConsumer from an existing ConsumerMessages and Consumer. + * + * This constructor is useful when you want to control the consumer lifecycle + * separately from the consumer. + * + * @category constructors * @since 0.1.0 */ -export type ConsumersTypeId = typeof ConsumersTypeId +export const fromConsumerMessages = ( + consumerMessages: JetStreamConsumerMessages.ConsumerMessages, + natsConsumer: JetStreamConsumerMessages.InfoableConsumer, + options: JetStreamConsumerOptions = {} +): Effect.Effect< + JetStreamConsumer, + NATSError.NATSConnectionError, + NATSConnection.NATSConnection +> => + Effect.gen(function*() { + const connection = yield* NATSConnection.NATSConnection + const connectionInfo = yield* Option.match(connection.info, { + onNone: () => Effect.fail(new NATSError.NATSConnectionError({ reason: "Connection info not available" })), + onSome: Effect.succeed + }) + + const consumer: JetStreamConsumer = { + [TypeId]: TypeId, + [Consumer.TypeId]: Consumer.TypeId, + serve: serveLayer(consumerMessages, connectionInfo, options), + serveEffect: serveEffect(consumerMessages, connectionInfo, options), + healthCheck: healthCheck(natsConsumer) + } + + return consumer + }) /** - * Represents consumers API + * Create a JetStreamConsumer from an existing Consumer. * - * @category models + * This is a convenience constructor that internally calls `consume()` on the consumer. + * + * @category constructors * @since 0.1.0 */ -export interface Consumers { - readonly [ConsumersTypeId]: ConsumersTypeId - readonly get: ( - ...args: Parameters - ) => Effect.Effect - readonly getConsumerFromInfo: ( - ...args: Parameters - ) => Effect.Effect - readonly getPushConsumer: ( - ...args: Parameters - ) => Effect.Effect - readonly getBoundPushConsumer: ( - ...args: Parameters - ) => Effect.Effect - - /** @internal */ - readonly consumers: JetStream.Consumers -} - -/** @internal */ -export const makeConsumers = (consumers: JetStream.Consumers): Consumers => ({ - [ConsumersTypeId]: ConsumersTypeId, - get: (...args) => wrapAsync(() => consumers.get(...args), "Failed to get consumer").pipe(Effect.map(makeConsumer)), - getConsumerFromInfo: (...args) => - wrap(() => (consumers.getConsumerFromInfo(...args)), "Failed to get consumer from info").pipe( - Effect.map(makeConsumer) - ), - getPushConsumer: (...args) => - wrapAsync(() => consumers.getPushConsumer(...args), "Failed to get push consumer").pipe( - Effect.map(makePushConsumer) - ), - getBoundPushConsumer: (...args) => - wrapAsync(() => consumers.getBoundPushConsumer(...args), "Failed to get bound push consumer").pipe( - Effect.map(makePushConsumer) - ), - consumers -}) +export const fromConsumer = ( + natsConsumer: JetStreamConsumerMessages.Consumer, + options: JetStreamConsumerOptions = {}, + consumeOptions?: Parameters[0] +): Effect.Effect< + JetStreamConsumer, + NATSError.JetStreamConsumerError | NATSError.NATSConnectionError, + NATSConnection.NATSConnection +> => + Effect.gen(function*() { + const consumerMessages = yield* natsConsumer.consume(consumeOptions) + return yield* fromConsumerMessages(consumerMessages, natsConsumer, options) + }) diff --git a/packages/nats/src/JetStreamConsumerMessages.ts b/packages/nats/src/JetStreamConsumerMessages.ts new file mode 100644 index 0000000..112e497 --- /dev/null +++ b/packages/nats/src/JetStreamConsumerMessages.ts @@ -0,0 +1,366 @@ +/** + * @since 0.1.0 + */ +import type * as JetStream from "@nats-io/jetstream" +import * as Effect from "effect/Effect" +import * as Option from "effect/Option" +import * as Stream from "effect/Stream" +import * as utils from "./internal/utils.js" +import * as JetStreamMessage from "./JetStreamMessage.js" +import * as NATSError from "./NATSError.js" +import * as NATSQueuedIterator from "./NATSQueuedIterator.js" + +const wrap = utils.wrap(NATSError.JetStreamConsumerError) +const wrapAsync = utils.wrapAsync(NATSError.JetStreamConsumerError) + +/** + * @category type ids + * @since 0.1.0 + */ +export const CloseTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Close") + +/** + * @category type ids + * @since 0.1.0 + */ +export type CloseTypeId = typeof CloseTypeId + +/** + * Represents closeable resources + * + * @category models + * @since 0.1.0 + */ +export interface Close { + readonly [CloseTypeId]: CloseTypeId + readonly close: Effect.Effect + readonly closed: Effect.Effect +} + +/** @internal */ +export const makeClose = (closeable: JetStream.Close): Close => ({ + [CloseTypeId]: CloseTypeId, + close: wrapAsync(() => closeable.close(), "Failed to close resource").pipe( + Effect.flatMap((result) => + result instanceof Error + ? Effect.fail(new NATSError.JetStreamConsumerError({ reason: "Failed to close resource", cause: result })) + : Effect.void + ) + ), + closed: wrapAsync(() => closeable.closed(), "Failed to check closed status").pipe( + Effect.flatMap((result) => + result instanceof Error + ? Effect.fail(new NATSError.JetStreamConsumerError({ reason: "Resource closed with error", cause: result })) + : Effect.void + ) + ) +}) + +/** + * @category type ids + * @since 0.1.0 + */ +export const ConsumerMessagesTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ConsumerMessages") + +/** + * @category type ids + * @since 0.1.0 + */ +export type ConsumerMessagesTypeId = typeof ConsumerMessagesTypeId + +/** + * Represents consumer messages + * + * @category models + * @since 0.1.0 + */ +export interface ConsumerMessages extends + Omit< + NATSQueuedIterator.NATSQueuedIterator, + "iterator" + >, + Close +{ + readonly [ConsumerMessagesTypeId]: ConsumerMessagesTypeId + readonly status: Effect.Effect< + Stream.Stream, + NATSError.JetStreamConsumerError + > + + /** @internal */ + readonly consumerMessages: JetStream.ConsumerMessages +} + +/** @internal */ +export const makeConsumerMessages = (consumerMessages: JetStream.ConsumerMessages): ConsumerMessages => { + const qi = NATSQueuedIterator.make(NATSError.JetStreamConsumerError)(consumerMessages) + const close = makeClose(consumerMessages) + return { + ...qi, + ...close, + [ConsumerMessagesTypeId]: ConsumerMessagesTypeId, + stream: Stream.map(qi.stream, JetStreamMessage.make), + status: wrap( + () => + Stream.fromAsyncIterable( + consumerMessages.status(), + (error) => + new NATSError.JetStreamConsumerError({ + reason: "An error occurred in consumer status async iterable", + cause: error + }) + ), + "Failed to get status stream" + ), + consumerMessages + } +} + +/** + * @category type ids + * @since 0.1.0 + */ +export const ConsumerKindTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ConsumerKind") + +/** + * @category type ids + * @since 0.1.0 + */ +export type ConsumerKindTypeId = typeof ConsumerKindTypeId + +/** + * Represents consumer kind + * + * @category models + * @since 0.1.0 + */ +export interface ConsumerKind { + readonly [ConsumerKindTypeId]: ConsumerKindTypeId + readonly isPullConsumer: Effect.Effect + readonly isPushConsumer: Effect.Effect +} + +/** + * @category type ids + * @since 0.1.0 + */ +export const ExportedConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/ExportedConsumer") + +/** + * @category type ids + * @since 0.1.0 + */ +export type ExportedConsumerTypeId = typeof ExportedConsumerTypeId + +/** + * Represents an exported consumer + * + * @category models + * @since 0.1.0 + */ +export interface ExportedConsumer extends ConsumerKind { + readonly [ExportedConsumerTypeId]: ExportedConsumerTypeId + readonly next: ( + ...args: Parameters + ) => Effect.Effect, NATSError.JetStreamConsumerError> + readonly fetch: ( + ...args: Parameters + ) => Effect.Effect + readonly consume: ( + ...args: Parameters + ) => Effect.Effect +} + +/** + * @category type ids + * @since 0.1.0 + */ +export const InfoableConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/InfoableConsumer") + +/** + * @category type ids + * @since 0.1.0 + */ +export type InfoableConsumerTypeId = typeof InfoableConsumerTypeId + +/** + * Represents an infoable consumer + * + * @category models + * @since 0.1.0 + */ +export interface InfoableConsumer { + readonly [InfoableConsumerTypeId]: InfoableConsumerTypeId + readonly info: ( + ...args: Parameters + ) => Effect.Effect +} + +/** + * @category type ids + * @since 0.1.0 + */ +export const DeleteableConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/DeleteableConsumer") + +/** + * @category type ids + * @since 0.1.0 + */ +export type DeleteableConsumerTypeId = typeof DeleteableConsumerTypeId + +/** + * Represents a deleteable consumer + * + * @category models + * @since 0.1.0 + */ +export interface DeleteableConsumer { + readonly [DeleteableConsumerTypeId]: DeleteableConsumerTypeId + readonly delete: Effect.Effect +} + +/** + * @category type ids + * @since 0.1.0 + */ +export const ConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Consumer") + +/** + * @category type ids + * @since 0.1.0 + */ +export type ConsumerTypeId = typeof ConsumerTypeId + +/** + * Represents a consumer + * + * @category models + * @since 0.1.0 + */ +export interface Consumer extends ExportedConsumer, InfoableConsumer, DeleteableConsumer { + readonly [ConsumerTypeId]: ConsumerTypeId + + /** @internal */ + readonly consumer: JetStream.Consumer +} + +/** @internal */ +export const makeConsumer = (consumer: JetStream.Consumer): Consumer => ({ + [ConsumerTypeId]: ConsumerTypeId, + [ExportedConsumerTypeId]: ExportedConsumerTypeId, + [ConsumerKindTypeId]: ConsumerKindTypeId, + [InfoableConsumerTypeId]: InfoableConsumerTypeId, + [DeleteableConsumerTypeId]: DeleteableConsumerTypeId, + isPullConsumer: wrap(() => consumer.isPullConsumer(), "Failed to check if consumer is pull consumer"), + isPushConsumer: wrap(() => consumer.isPushConsumer(), "Failed to check if consumer is push consumer"), + next: (...args) => + wrapAsync(() => consumer.next(...args), "Failed to get next message").pipe( + Effect.map((msg) => Option.fromNullable(msg ? JetStreamMessage.make(msg) : null)) + ), + fetch: (...args) => + wrapAsync(() => consumer.fetch(...args), "Failed to fetch messages").pipe(Effect.map(makeConsumerMessages)), + consume: (...args) => + wrapAsync(() => consumer.consume(...args), "Failed to consume messages").pipe(Effect.map(makeConsumerMessages)), + info: (...args) => wrapAsync(() => consumer.info(...args), "Failed to get consumer info"), + delete: wrapAsync(() => consumer.delete(), "Failed to delete consumer"), + consumer +}) + +/** + * @category type ids + * @since 0.1.0 + */ +export const PushConsumerTypeId: unique symbol = Symbol.for("@effect-messaging/nats/PushConsumer") + +/** + * @category type ids + * @since 0.1.0 + */ +export type PushConsumerTypeId = typeof PushConsumerTypeId + +/** + * Represents a push consumer + * + * @category models + * @since 0.1.0 + */ +export interface PushConsumer extends InfoableConsumer, DeleteableConsumer, ConsumerKind { + readonly [PushConsumerTypeId]: PushConsumerTypeId + readonly consume: ( + ...args: Parameters + ) => Effect.Effect + + /** @internal */ + readonly pushConsumer: JetStream.PushConsumer +} + +/** @internal */ +export const makePushConsumer = (pushConsumer: JetStream.PushConsumer): PushConsumer => ({ + [PushConsumerTypeId]: PushConsumerTypeId, + [ConsumerKindTypeId]: ConsumerKindTypeId, + [InfoableConsumerTypeId]: InfoableConsumerTypeId, + [DeleteableConsumerTypeId]: DeleteableConsumerTypeId, + isPullConsumer: wrap(() => pushConsumer.isPullConsumer(), "Failed to check if consumer is pull consumer"), + isPushConsumer: wrap(() => pushConsumer.isPushConsumer(), "Failed to check if consumer is push consumer"), + consume: (...args) => + wrapAsync(() => pushConsumer.consume(...args), "Failed to consume messages").pipe(Effect.map(makeConsumerMessages)), + info: (...args) => wrapAsync(() => pushConsumer.info(...args), "Failed to get consumer info"), + delete: wrapAsync(() => pushConsumer.delete(), "Failed to delete consumer"), + pushConsumer +}) + +/** + * @category type ids + * @since 0.1.0 + */ +export const ConsumersTypeId: unique symbol = Symbol.for("@effect-messaging/nats/Consumers") + +/** + * @category type ids + * @since 0.1.0 + */ +export type ConsumersTypeId = typeof ConsumersTypeId + +/** + * Represents consumers API + * + * @category models + * @since 0.1.0 + */ +export interface Consumers { + readonly [ConsumersTypeId]: ConsumersTypeId + readonly get: ( + ...args: Parameters + ) => Effect.Effect + readonly getConsumerFromInfo: ( + ...args: Parameters + ) => Effect.Effect + readonly getPushConsumer: ( + ...args: Parameters + ) => Effect.Effect + readonly getBoundPushConsumer: ( + ...args: Parameters + ) => Effect.Effect + + /** @internal */ + readonly consumers: JetStream.Consumers +} + +/** @internal */ +export const makeConsumers = (consumers: JetStream.Consumers): Consumers => ({ + [ConsumersTypeId]: ConsumersTypeId, + get: (...args) => wrapAsync(() => consumers.get(...args), "Failed to get consumer").pipe(Effect.map(makeConsumer)), + getConsumerFromInfo: (...args) => + wrap(() => (consumers.getConsumerFromInfo(...args)), "Failed to get consumer from info").pipe( + Effect.map(makeConsumer) + ), + getPushConsumer: (...args) => + wrapAsync(() => consumers.getPushConsumer(...args), "Failed to get push consumer").pipe( + Effect.map(makePushConsumer) + ), + getBoundPushConsumer: (...args) => + wrapAsync(() => consumers.getBoundPushConsumer(...args), "Failed to get bound push consumer").pipe( + Effect.map(makePushConsumer) + ), + consumers +}) diff --git a/packages/nats/src/JetStreamSubscriberResponse.ts b/packages/nats/src/JetStreamConsumerResponse.ts similarity index 76% rename from packages/nats/src/JetStreamSubscriberResponse.ts rename to packages/nats/src/JetStreamConsumerResponse.ts index c0e05d7..d76a246 100644 --- a/packages/nats/src/JetStreamSubscriberResponse.ts +++ b/packages/nats/src/JetStreamConsumerResponse.ts @@ -6,7 +6,7 @@ * @category type ids * @since 0.7.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamSubscriberResponse") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamConsumerResponse") /** * @category type ids @@ -63,7 +63,7 @@ export interface Term { * @category models * @since 0.7.0 */ -export type JetStreamSubscriberResponse = Ack | Nak | Term +export type JetStreamConsumerResponse = Ack | Nak | Term class AckImpl implements Ack { readonly [TypeId]: TypeId = TypeId @@ -88,23 +88,23 @@ class TermImpl implements Term { * @category constructors * @since 0.7.0 */ -export const ack = (): JetStreamSubscriberResponse => new AckImpl() +export const ack = (): JetStreamConsumerResponse => new AckImpl() /** * @category constructors * @since 0.7.0 */ -export const nak = (options?: NakOptions): JetStreamSubscriberResponse => new NakImpl(options?.millis) +export const nak = (options?: NakOptions): JetStreamConsumerResponse => new NakImpl(options?.millis) /** * @category constructors * @since 0.7.0 */ -export const term = (options?: TermOptions): JetStreamSubscriberResponse => new TermImpl(options?.reason) +export const term = (options?: TermOptions): JetStreamConsumerResponse => new TermImpl(options?.reason) /** * @category guards * @since 0.7.0 */ -export const isJetStreamSubscriberResponse = (u: unknown): u is JetStreamSubscriberResponse => +export const isJetStreamConsumerResponse = (u: unknown): u is JetStreamConsumerResponse => typeof u === "object" && u !== null && TypeId in u diff --git a/packages/nats/src/JetStreamPublisher.ts b/packages/nats/src/JetStreamProducer.ts similarity index 81% rename from packages/nats/src/JetStreamPublisher.ts rename to packages/nats/src/JetStreamProducer.ts index 3d4ff94..ea069d6 100644 --- a/packages/nats/src/JetStreamPublisher.ts +++ b/packages/nats/src/JetStreamProducer.ts @@ -1,8 +1,8 @@ /** * @since 0.1.0 */ -import * as Publisher from "@effect-messaging/core/Publisher" -import * as PublisherError from "@effect-messaging/core/PublisherError" +import * as Producer from "@effect-messaging/core/Producer" +import * as ProducerError from "@effect-messaging/core/ProducerError" import type * as JetStream from "@nats-io/jetstream" import type * as NATSCore from "@nats-io/nats-core" import * as Effect from "effect/Effect" @@ -18,7 +18,7 @@ import * as NATSHeaders from "./NATSHeaders.js" * @category type ids * @since 0.1.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamPublisher") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamProducer") /** * @category type ids @@ -40,7 +40,7 @@ export interface JetStreamPublishMessage { * @category models * @since 0.1.0 */ -export interface JetStreamPublisher extends Publisher.Publisher { +export interface JetStreamProducer extends Producer.Producer { readonly [TypeId]: TypeId } @@ -73,7 +73,7 @@ const publish = ( connectionInfo: NATSCore.ServerInfo, retrySchedule: Schedule.Schedule ) => -(message: JetStreamPublishMessage): Effect.Effect => +(message: JetStreamPublishMessage): Effect.Effect => Effect.useSpan( `nats.publish ${message.subject}`, { @@ -94,8 +94,7 @@ const publish = ( Effect.retry(retrySchedule), Effect.catchTag( "JetStreamClientError", - (error) => - Effect.fail(new PublisherError.PublisherError({ reason: "Failed to publish message", cause: error })) + (error) => Effect.fail(new ProducerError.ProducerError({ reason: "Failed to publish message", cause: error })) ) ) ) @@ -104,7 +103,7 @@ const publish = ( * @category constructors * @since 0.1.0 */ -export interface JetStreamPublisherConfig { +export interface JetStreamProducerConfig { readonly retrySchedule?: Schedule.Schedule } @@ -113,9 +112,9 @@ export interface JetStreamPublisherConfig { * @since 0.1.0 */ export const make = ( - config?: JetStreamPublisherConfig + config?: JetStreamProducerConfig ): Effect.Effect< - JetStreamPublisher, + JetStreamProducer, NATSError.JetStreamClientError | NATSError.NATSConnectionError, JetStreamClient.JetStreamClient | NATSConnection.NATSConnection > => @@ -129,11 +128,11 @@ export const make = ( onSome: Effect.succeed }) - const publisher: JetStreamPublisher = { + const producer: JetStreamProducer = { [TypeId]: TypeId, - [Publisher.TypeId]: Publisher.TypeId, - publish: publish(client, connectionInfo, config?.retrySchedule ?? Schedule.stop) + [Producer.TypeId]: Producer.TypeId, + send: publish(client, connectionInfo, config?.retrySchedule ?? Schedule.stop) } - return publisher + return producer }) diff --git a/packages/nats/src/JetStreamStream.ts b/packages/nats/src/JetStreamStream.ts index bbfbe39..ce66fba 100644 --- a/packages/nats/src/JetStreamStream.ts +++ b/packages/nats/src/JetStreamStream.ts @@ -5,7 +5,7 @@ import type * as JetStream from "@nats-io/jetstream" import * as Effect from "effect/Effect" import * as Option from "effect/Option" import * as utils from "./internal/utils.js" -import * as JetStreamConsumers from "./JetStreamConsumer.js" +import * as JetStreamConsumerMessages from "./JetStreamConsumerMessages.js" import * as JetStreamStoredMessage from "./JetStreamStoredMessage.js" import * as NATSError from "./NATSError.js" @@ -48,7 +48,7 @@ export interface JetStreamStream { readonly best: Effect.Effect readonly getConsumer: ( ...args: Parameters - ) => Effect.Effect + ) => Effect.Effect /** @internal */ readonly stream: JetStream.Stream @@ -69,7 +69,7 @@ export const makeJetStreamStream = (stream: JetStream.Stream): JetStreamStream = best: wrapAsync(() => stream.best(), "Failed to get best stream").pipe(Effect.map(makeJetStreamStream)), getConsumer: (...args) => wrapAsync(() => stream.getConsumer(...args), "Failed to get consumer").pipe( - Effect.map(JetStreamConsumers.makeConsumer) + Effect.map(JetStreamConsumerMessages.makeConsumer) ), stream }) diff --git a/packages/nats/src/JetStreamSubscriber.ts b/packages/nats/src/JetStreamSubscriber.ts deleted file mode 100644 index 618a617..0000000 --- a/packages/nats/src/JetStreamSubscriber.ts +++ /dev/null @@ -1,234 +0,0 @@ -/** - * @since 0.1.0 - */ -import * as Subscriber from "@effect-messaging/core/Subscriber" -import type * as SubscriberApp from "@effect-messaging/core/SubscriberApp" -import * as SubscriberError from "@effect-messaging/core/SubscriberError" -import type * as NATSCore from "@nats-io/nats-core" -import * as Cause from "effect/Cause" -import type * as Duration from "effect/Duration" -import * as Effect from "effect/Effect" -import * as Function from "effect/Function" -import * as Match from "effect/Match" -import * as Option from "effect/Option" -import * as Predicate from "effect/Predicate" -import * as Stream from "effect/Stream" -import type * as JetStreamConsumer from "./JetStreamConsumer.js" -import * as JetStreamMessage from "./JetStreamMessage.js" -import type * as JetStreamSubscriberResponse from "./JetStreamSubscriberResponse.js" -import * as NATSConnection from "./NATSConnection.js" -import * as NATSError from "./NATSError.js" -import * as NATSHeaders from "./NATSHeaders.js" - -/** - * @category type ids - * @since 0.1.0 - */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/JetStreamSubscriber") - -/** - * @category type ids - * @since 0.1.0 - */ -export type TypeId = typeof TypeId - -/** - * @category models - * @since 0.7.0 - */ -export type JetStreamSubscriberApp = SubscriberApp.SubscriberApp< - JetStreamSubscriberResponse.JetStreamSubscriberResponse, - JetStreamMessage.JetStreamMessage, - E, - R -> - -/** - * @category models - * @since 0.1.0 - */ -export interface JetStreamSubscriber - extends - Subscriber.Subscriber -{ - readonly [TypeId]: TypeId -} - -/** - * @category models - * @since 0.1.0 - */ -export interface JetStreamSubscriberOptions { - uninterruptible?: boolean - handlerTimeout?: Duration.DurationInput -} - -const ATTR_SERVER_ADDRESS = "server.address" as const -const ATTR_SERVER_PORT = "server.port" as const -const ATTR_MESSAGING_DESTINATION_NAME = "messaging.destination.name" as const -const ATTR_MESSAGING_OPERATION_NAME = "messaging.operation.name" as const -const ATTR_MESSAGING_OPERATION_TYPE = "messaging.operation.type" as const -const ATTR_MESSAGING_SYSTEM = "messaging.system" as const -const ATTR_MESSAGING_MESSAGE_ID = "messaging.message.id" as const -const ATTR_MESSAGING_NATS_STREAM = "messaging.nats.stream" as const -const ATTR_MESSAGING_NATS_CONSUMER = "messaging.nats.consumer" as const -const ATTR_MESSAGING_NATS_SEQUENCE_STREAM = "messaging.nats.sequence.stream" as const -const ATTR_MESSAGING_NATS_SEQUENCE_CONSUMER = "messaging.nats.sequence.consumer" as const - -/** @internal */ -const subscribe = ( - consumerMessages: JetStreamConsumer.ConsumerMessages, - connectionInfo: NATSCore.ServerInfo, - options: JetStreamSubscriberOptions -) => -( - app: JetStreamSubscriberApp -): Effect.Effect> => - consumerMessages.stream.pipe( - Stream.runForEach((message) => - Effect.fork( - Effect.useSpan( - `nats.consume ${message.subject}`, - { - parent: Option.getOrUndefined(NATSHeaders.decodeTraceContextOptional(message.headers)), - kind: "consumer", - captureStackTrace: false, - attributes: { - [ATTR_SERVER_ADDRESS]: connectionInfo.host, - [ATTR_SERVER_PORT]: connectionInfo.port, - [ATTR_MESSAGING_SYSTEM]: "nats", - [ATTR_MESSAGING_OPERATION_TYPE]: "receive", - [ATTR_MESSAGING_DESTINATION_NAME]: message.subject, - [ATTR_MESSAGING_MESSAGE_ID]: message.seq, - [ATTR_MESSAGING_NATS_STREAM]: message.info.stream, - [ATTR_MESSAGING_NATS_CONSUMER]: message.info.consumer, - [ATTR_MESSAGING_NATS_SEQUENCE_STREAM]: message.info.streamSequence, - [ATTR_MESSAGING_NATS_SEQUENCE_CONSUMER]: message.info.deliverySequence - } - }, - (span) => - Effect.gen(function*() { - yield* Effect.logDebug(`nats.consume ${message.subject}`) - const response = yield* app.pipe( - options.handlerTimeout - ? Effect.timeoutFail({ - duration: options.handlerTimeout, - onTimeout: () => - new SubscriberError.SubscriberError({ reason: "JetStreamSubscriber: handler timed out" }) - }) - : Function.identity - ) - yield* Match.valueTags(response, { - Ack: () => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "ack") - yield* message.ack - }), - Nak: (r) => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nak") - yield* message.nak(r.millis) - }), - Term: (r) => - Effect.gen(function*() { - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "term") - yield* message.term(r.reason) - }) - }) - }).pipe( - Effect.provide(JetStreamMessage.layer(message)), - Effect.tapErrorCause((cause) => - Effect.gen(function*() { - yield* Effect.logError(Cause.pretty(cause)) - span.attribute(ATTR_MESSAGING_OPERATION_NAME, "nak") - span.attribute( - "error.type", - String(Cause.squashWith(cause, (_) => - Predicate.hasProperty(_, "_tag") ? _._tag : _ instanceof Error ? _.name : `${_}`)) - ) - span.attribute("error.stack", Cause.pretty(cause)) - span.attribute( - "error.message", - String(Cause.squashWith(cause, (_) => - Predicate.hasProperty(_, "reason") ? _.reason : _ instanceof Error ? _.message : `${_}`)) - ) - yield* message.nak() - }) - ), - options.uninterruptible ? Effect.uninterruptible : Effect.interruptible, - Effect.withParentSpan(span) - ) - ) - ) - ), - Effect.mapError((error) => - new SubscriberError.SubscriberError({ reason: "JetStreamSubscriber failed to subscribe", cause: error }) - ) - ) - -/** @internal */ -const healthCheck = ( - consumer: JetStreamConsumer.InfoableConsumer -): Effect.Effect => - consumer.info().pipe( - Effect.catchTag("JetStreamConsumerError", (error) => - new SubscriberError.SubscriberError({ reason: "Healthcheck failed", cause: error })), - Effect.asVoid - ) - -/** - * Create a JetStreamSubscriber from an existing ConsumerMessages and Consumer. - * - * This constructor is useful when you want to control the consumer lifecycle - * separately from the subscriber. - * - * @category constructors - * @since 0.1.0 - */ -export const fromConsumerMessages = ( - consumerMessages: JetStreamConsumer.ConsumerMessages, - consumer: JetStreamConsumer.InfoableConsumer, - options: JetStreamSubscriberOptions = {} -): Effect.Effect< - JetStreamSubscriber, - NATSError.NATSConnectionError, - NATSConnection.NATSConnection -> => - Effect.gen(function*() { - const connection = yield* NATSConnection.NATSConnection - const connectionInfo = yield* Option.match(connection.info, { - onNone: () => Effect.fail(new NATSError.NATSConnectionError({ reason: "Connection info not available" })), - onSome: Effect.succeed - }) - - const subscriber: JetStreamSubscriber = { - [TypeId]: TypeId, - [Subscriber.TypeId]: Subscriber.TypeId, - subscribe: subscribe(consumerMessages, connectionInfo, options), - healthCheck: healthCheck(consumer) - } - - return subscriber - }) - -/** - * Create a JetStreamSubscriber from an existing Consumer. - * - * This is a convenience constructor that internally calls `consume()` on the consumer. - * - * @category constructors - * @since 0.1.0 - */ -export const fromConsumer = ( - consumer: JetStreamConsumer.Consumer, - options: JetStreamSubscriberOptions = {}, - consumeOptions?: Parameters[0] -): Effect.Effect< - JetStreamSubscriber, - NATSError.JetStreamConsumerError | NATSError.NATSConnectionError, - NATSConnection.NATSConnection -> => - Effect.gen(function*() { - const consumerMessages = yield* consumer.consume(consumeOptions) - return yield* fromConsumerMessages(consumerMessages, consumer, options) - }) diff --git a/packages/nats/src/NATSSubscriber.ts b/packages/nats/src/NATSConsumer.ts similarity index 76% rename from packages/nats/src/NATSSubscriber.ts rename to packages/nats/src/NATSConsumer.ts index b278cb7..7b4046f 100644 --- a/packages/nats/src/NATSSubscriber.ts +++ b/packages/nats/src/NATSConsumer.ts @@ -1,13 +1,15 @@ /** * @since 0.3.0 */ -import * as Subscriber from "@effect-messaging/core/Subscriber" -import * as SubscriberError from "@effect-messaging/core/SubscriberError" +import * as Consumer from "@effect-messaging/core/Consumer" +import type * as ConsumerApp from "@effect-messaging/core/ConsumerApp" +import * as ConsumerError from "@effect-messaging/core/ConsumerError" import type * as NATSCore from "@nats-io/nats-core" import * as Cause from "effect/Cause" import type * as Duration from "effect/Duration" import * as Effect from "effect/Effect" import * as Function from "effect/Function" +import * as Layer from "effect/Layer" import * as Option from "effect/Option" import * as Predicate from "effect/Predicate" import * as Stream from "effect/Stream" @@ -21,7 +23,7 @@ import type * as NATSSubscription from "./NATSSubscription.js" * @category type ids * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/NATSSubscriber") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/NATSConsumer") /** * @category type ids @@ -33,7 +35,13 @@ export type TypeId = typeof TypeId * @category models * @since 0.3.0 */ -export interface NATSSubscriber extends Subscriber.Subscriber { +export type NATSConsumerApp = ConsumerApp.ConsumerApp + +/** + * @category models + * @since 0.3.0 + */ +export interface NATSConsumer extends Consumer.Consumer { readonly [TypeId]: TypeId } @@ -41,7 +49,7 @@ export interface NATSSubscriber extends Subscriber.Subscriber -( - handler: Effect.Effect -): Effect.Effect> => +(app: NATSConsumerApp) => subscription.stream.pipe( Stream.runForEach((message) => Effect.fork( @@ -84,12 +90,11 @@ const subscribe = ( (span) => Effect.gen(function*() { yield* Effect.logDebug(`nats.consume ${message.subject}`) - yield* handler.pipe( + yield* app.pipe( options.handlerTimeout ? Effect.timeoutFail({ duration: options.handlerTimeout, - onTimeout: () => - new SubscriberError.SubscriberError({ reason: "NATSSubscriber: handler timed out" }) + onTimeout: () => new ConsumerError.ConsumerError({ reason: "NATSConsumer: handler timed out" }) }) : Function.identity ) @@ -125,26 +130,36 @@ const subscribe = ( ) ), Effect.mapError((error) => - new SubscriberError.SubscriberError({ reason: "NATSSubscriber failed to subscribe", cause: error }) + new ConsumerError.ConsumerError({ reason: "NATSConsumer failed to subscribe", cause: error }) ) ) +/** @internal */ +const serveLayer = ( + subscription: NATSSubscription.NATSSubscription, + connectionInfo: NATSCore.ServerInfo, + options: NATSConsumerOptions +) => +( + handler: Effect.Effect +) => Layer.scopedDiscard(serveEffect(subscription, connectionInfo, options)(handler)) + /** @internal */ const healthCheck = ( subscription: NATSSubscription.NATSSubscription -): Effect.Effect => +): Effect.Effect => subscription.isClosed.pipe( Effect.flatMap((isClosed) => isClosed - ? Effect.fail(new SubscriberError.SubscriberError({ reason: "Subscription is closed" })) + ? Effect.fail(new ConsumerError.ConsumerError({ reason: "Subscription is closed" })) : Effect.void ), Effect.catchTag("NATSSubscriptionError", (error) => - new SubscriberError.SubscriberError({ reason: "Healthcheck failed", cause: error })) + new ConsumerError.ConsumerError({ reason: "Healthcheck failed", cause: error })) ) /** - * Create a NATSSubscriber from an existing NATSSubscription. + * Create a NATSConsumer from an existing NATSSubscription. * * Note: NATS Core subscriptions are fire-and-forget. Messages are not persisted * and there is no acknowledgment mechanism. If the handler fails or times out, @@ -155,9 +170,9 @@ const healthCheck = ( */ export const fromSubscription = ( subscription: NATSSubscription.NATSSubscription, - options: NATSSubscriberOptions = {} + options: NATSConsumerOptions = {} ): Effect.Effect< - NATSSubscriber, + NATSConsumer, NATSError.NATSConnectionError, NATSConnection.NATSConnection > => @@ -168,18 +183,19 @@ export const fromSubscription = ( onSome: Effect.succeed }) - const subscriber: NATSSubscriber = { + const consumer: NATSConsumer = { [TypeId]: TypeId, - [Subscriber.TypeId]: Subscriber.TypeId, - subscribe: subscribe(subscription, connectionInfo, options), + [Consumer.TypeId]: Consumer.TypeId, + serve: serveLayer(subscription, connectionInfo, options), + serveEffect: serveEffect(subscription, connectionInfo, options), healthCheck: healthCheck(subscription) } - return subscriber + return consumer }) /** - * Create a NATSSubscriber by subscribing to a subject. + * Create a NATSConsumer by subscribing to a subject. * * This is a convenience constructor that internally calls `subscribe()` on the connection. * @@ -193,9 +209,9 @@ export const fromSubscription = ( export const make = ( subject: string, subscriptionOptions?: NATSCore.SubscriptionOptions, - options: NATSSubscriberOptions = {} + options: NATSConsumerOptions = {} ): Effect.Effect< - NATSSubscriber, + NATSConsumer, NATSError.NATSConnectionError, NATSConnection.NATSConnection > => diff --git a/packages/nats/src/NATSPublisher.ts b/packages/nats/src/NATSProducer.ts similarity index 80% rename from packages/nats/src/NATSPublisher.ts rename to packages/nats/src/NATSProducer.ts index 7a91117..eb69c63 100644 --- a/packages/nats/src/NATSPublisher.ts +++ b/packages/nats/src/NATSProducer.ts @@ -1,8 +1,8 @@ /** * @since 0.3.0 */ -import * as Publisher from "@effect-messaging/core/Publisher" -import * as PublisherError from "@effect-messaging/core/PublisherError" +import * as Producer from "@effect-messaging/core/Producer" +import * as ProducerError from "@effect-messaging/core/ProducerError" import type * as NATSCore from "@nats-io/nats-core" import * as Effect from "effect/Effect" import * as Option from "effect/Option" @@ -16,7 +16,7 @@ import * as NATSHeaders from "./NATSHeaders.js" * @category type ids * @since 0.3.0 */ -export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/NATSPublisher") +export const TypeId: unique symbol = Symbol.for("@effect-messaging/nats/NATSProducer") /** * @category type ids @@ -38,7 +38,7 @@ export interface NATSPublishMessage { * @category models * @since 0.3.0 */ -export interface NATSPublisher extends Publisher.Publisher { +export interface NATSProducer extends Producer.Producer { readonly [TypeId]: TypeId } @@ -70,7 +70,7 @@ const publish = ( connectionInfo: NATSCore.ServerInfo, retrySchedule: Schedule.Schedule ) => -(message: NATSPublishMessage): Effect.Effect => +(message: NATSPublishMessage): Effect.Effect => Effect.useSpan( `nats.publish ${message.subject}`, { @@ -90,8 +90,7 @@ const publish = ( Effect.retry(retrySchedule), Effect.catchTag( "NATSConnectionError", - (error) => - Effect.fail(new PublisherError.PublisherError({ reason: "Failed to publish message", cause: error })) + (error) => Effect.fail(new ProducerError.ProducerError({ reason: "Failed to publish message", cause: error })) ) ) ) @@ -100,7 +99,7 @@ const publish = ( * @category constructors * @since 0.3.0 */ -export interface NATSPublisherConfig { +export interface NATSProducerConfig { readonly retrySchedule?: Schedule.Schedule } @@ -109,9 +108,9 @@ export interface NATSPublisherConfig { * @since 0.3.0 */ export const make = ( - config?: NATSPublisherConfig + config?: NATSProducerConfig ): Effect.Effect< - NATSPublisher, + NATSProducer, NATSError.NATSConnectionError, NATSConnection.NATSConnection > => @@ -124,11 +123,11 @@ export const make = ( onSome: Effect.succeed }) - const publisher: NATSPublisher = { + const producer: NATSProducer = { [TypeId]: TypeId, - [Publisher.TypeId]: Publisher.TypeId, - publish: publish(connection, connectionInfo, config?.retrySchedule ?? Schedule.stop) + [Producer.TypeId]: Producer.TypeId, + send: publish(connection, connectionInfo, config?.retrySchedule ?? Schedule.stop) } - return publisher + return producer }) diff --git a/packages/nats/src/index.ts b/packages/nats/src/index.ts index 8c0bc16..6c75ca7 100644 --- a/packages/nats/src/index.ts +++ b/packages/nats/src/index.ts @@ -18,6 +18,16 @@ export * as JetStreamConsumer from "./JetStreamConsumer.js" */ export * as JetStreamConsumerAPI from "./JetStreamConsumerAPI.js" +/** + * @since 0.1.0 + */ +export * as JetStreamConsumerMessages from "./JetStreamConsumerMessages.js" + +/** + * @since 0.7.0 + */ +export * as JetStreamConsumerResponse from "./JetStreamConsumerResponse.js" + /** * @since 0.1.0 */ @@ -41,7 +51,7 @@ export * as JetStreamMessage from "./JetStreamMessage.js" /** * @since 0.1.0 */ -export * as JetStreamPublisher from "./JetStreamPublisher.js" +export * as JetStreamProducer from "./JetStreamProducer.js" /** * @since 0.1.0 @@ -61,17 +71,12 @@ export * as JetStreamStreamAPI from "./JetStreamStreamAPI.js" /** * @since 0.1.0 */ -export * as JetStreamSubscriber from "./JetStreamSubscriber.js" - -/** - * @since 0.7.0 - */ -export * as JetStreamSubscriberResponse from "./JetStreamSubscriberResponse.js" +export * as NATSConnection from "./NATSConnection.js" /** - * @since 0.1.0 + * @since 0.3.0 */ -export * as NATSConnection from "./NATSConnection.js" +export * as NATSConsumer from "./NATSConsumer.js" /** * @since 0.1.0 @@ -91,18 +96,13 @@ export * as NATSMessage from "./NATSMessage.js" /** * @since 0.3.0 */ -export * as NATSPublisher from "./NATSPublisher.js" +export * as NATSProducer from "./NATSProducer.js" /** * @since 0.1.0 */ export * as NATSQueuedIterator from "./NATSQueuedIterator.js" -/** - * @since 0.3.0 - */ -export * as NATSSubscriber from "./NATSSubscriber.js" - /** * @since 0.1.0 */ diff --git a/packages/nats/test/JetStreamSubscriber.test.ts b/packages/nats/test/JetStreamConsumer.test.ts similarity index 78% rename from packages/nats/test/JetStreamSubscriber.test.ts rename to packages/nats/test/JetStreamConsumer.test.ts index 335cb36..012d66d 100644 --- a/packages/nats/test/JetStreamSubscriber.test.ts +++ b/packages/nats/test/JetStreamConsumer.test.ts @@ -1,28 +1,28 @@ import type { Mock } from "@effect/vitest" import { describe, expect, it, vi } from "@effect/vitest" -import { Effect, Schedule, TestServices } from "effect" +import { Effect, Layer, Schedule, TestServices } from "effect" import * as JetStreamClient from "../src/JetStreamClient.js" +import * as JetStreamConsumer from "../src/JetStreamConsumer.js" +import * as JetStreamConsumerResponse from "../src/JetStreamConsumerResponse.js" import * as JetStreamMessage from "../src/JetStreamMessage.js" -import * as JetStreamPublisher from "../src/JetStreamPublisher.js" -import * as JetStreamSubscriber from "../src/JetStreamSubscriber.js" -import * as JetStreamSubscriberResponse from "../src/JetStreamSubscriberResponse.js" +import * as JetStreamProducer from "../src/JetStreamProducer.js" import { makeTestConsumer, makeTestStream, purgeTestStream, testJetStream } from "./dependencies.js" // Use unique names for this test file to avoid conflicts with other test files -const TEST_STREAM = "SUBSCRIBER_TEST_STREAM" -const TEST_CONSUMER = "SUBSCRIBER_TEST_CONSUMER" -const TEST_SUBJECT = "subscriber.test.subject" +const TEST_STREAM = "CONSUMER_TEST_STREAM" +const TEST_CONSUMER = "CONSUMER_TEST_CONSUMER" +const TEST_SUBJECT = "consumer.test.subject" const publishAndAssertConsume = ( - { content, onMessage, publisher, times }: { - publisher: JetStreamPublisher.JetStreamPublisher + { content, onMessage, producer, times }: { + producer: JetStreamProducer.JetStreamProducer onMessage: Mock<(message: JetStreamMessage.JetStreamMessage) => void> content: Uint8Array times: number } ) => Effect.gen(function*() { - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: content }) @@ -44,14 +44,14 @@ const setup = Effect.gen(function*() { yield* purgeTestStream(TEST_STREAM) }) -describe("JetStreamSubscriber", { sequential: true }, () => { - describe("subscribe", () => { +describe("JetStreamConsumer", { sequential: true }, () => { + describe("serve", () => { it.effect("Should consume published events", () => Effect.scoped( Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make({ + const producer = yield* JetStreamProducer.make({ retrySchedule: Schedule.exponential("100 millis", 1.5).pipe( Schedule.jittered, Schedule.intersect(Schedule.recurs(10)) @@ -59,21 +59,21 @@ describe("JetStreamSubscriber", { sequential: true }, () => { }) const client = yield* JetStreamClient.JetStreamClient - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) const onMessage = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() - // Start the subscription - yield* Effect.fork(subscriber.subscribe(Effect.gen(function*() { + // Start the subscription using Layer.launch + yield* Effect.fork(Layer.launch(consumer.serve(Effect.gen(function*() { const message = yield* JetStreamMessage.JetStreamConsumeMessage onMessage(message) - return JetStreamSubscriberResponse.ack() - }))) + return JetStreamConsumerResponse.ack() + })))) // Message 1 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 1"), times: 1 @@ -81,7 +81,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { // Message 2 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 2"), times: 2 @@ -89,7 +89,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { // Message 3 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 3"), times: 3 @@ -98,7 +98,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { ).pipe(Effect.provide(testJetStream), TestServices.provideLive)) }) - describe("interruptable subscribers", { sequential: true }, () => { + describe("interruptable consumers", { sequential: true }, () => { it.effect( "Should interrupt the handler if the subscription fiber is interrupted, and the message should be consumed again", () => @@ -106,7 +106,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() const onHandlingFinished = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() @@ -116,21 +116,21 @@ describe("JetStreamSubscriber", { sequential: true }, () => { onHandlingStarted(message) yield* Effect.sleep("500 millis") onHandlingFinished(message) - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) - yield* subscriber.subscribe(handler) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription const subscriptionFiber1 = yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will be interrupted") }) @@ -161,12 +161,12 @@ describe("JetStreamSubscriber", { sequential: true }, () => { { timeout: 15000 } ) - it.effect("Should not interrupt the handler if the subscriber is uninterruptible", () => + it.effect("Should not interrupt the handler if the consumer is uninterruptible", () => Effect.scoped( Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() const onHandlingFinished = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() @@ -176,21 +176,21 @@ describe("JetStreamSubscriber", { sequential: true }, () => { onHandlingStarted(message) yield* Effect.sleep("300 millis") onHandlingFinished(message) - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer, { uninterruptible: true }) - yield* subscriber.subscribe(handler) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer, { uninterruptible: true }) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription const subscriptionFiber1 = yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will NOT be interrupted") }) @@ -224,7 +224,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() const onHandlingFinished = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() @@ -243,23 +243,23 @@ describe("JetStreamSubscriber", { sequential: true }, () => { yield* Effect.sleep("50 millis") } onHandlingFinished(message) - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer, { + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer, { handlerTimeout: "300 millis" }) - yield* subscriber.subscribe(handler) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will timeout on first attempt") }) @@ -287,7 +287,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() const onHandlingFinished = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() @@ -304,21 +304,21 @@ describe("JetStreamSubscriber", { sequential: true }, () => { } onHandlingFinished(message) - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) - yield* subscriber.subscribe(handler) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message that will fail first time") }) @@ -339,7 +339,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() let attemptCount = 0 @@ -351,25 +351,25 @@ describe("JetStreamSubscriber", { sequential: true }, () => { if (attemptCount === 1) { // First attempt: nak with delay to trigger redelivery - return JetStreamSubscriberResponse.nak({ millis: 100 }) + return JetStreamConsumerResponse.nak({ millis: 100 }) } // Subsequent attempts: ack - return JetStreamSubscriberResponse.ack() + return JetStreamConsumerResponse.ack() }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) - yield* subscriber.subscribe(handler) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message that will be nacked first") }) @@ -387,7 +387,7 @@ describe("JetStreamSubscriber", { sequential: true }, () => { Effect.gen(function*() { yield* setup - const publisher = yield* JetStreamPublisher.make() + const producer = yield* JetStreamProducer.make() const onHandlingStarted = vi.fn<(message: JetStreamMessage.JetStreamMessage) => void>() @@ -396,21 +396,21 @@ describe("JetStreamSubscriber", { sequential: true }, () => { onHandlingStarted(message) // Terminate the message with a reason - message won't be redelivered - return JetStreamSubscriberResponse.term({ reason: "Intentionally terminated for testing" }) + return JetStreamConsumerResponse.term({ reason: "Intentionally terminated for testing" }) }) const client = yield* JetStreamClient.JetStreamClient const startSubscription = Effect.gen(function*() { - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) - yield* subscriber.subscribe(handler) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) + return yield* Layer.launch(consumer.serve(handler)) }) // Start the subscription yield* Effect.fork(startSubscription) - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message that will be terminated") }) @@ -435,11 +435,11 @@ describe("JetStreamSubscriber", { sequential: true }, () => { yield* setup const client = yield* JetStreamClient.JetStreamClient - const consumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) - const subscriber = yield* JetStreamSubscriber.fromConsumer(consumer) + const natsConsumer = yield* client.consumers.get(TEST_STREAM, TEST_CONSUMER) + const consumer = yield* JetStreamConsumer.fromConsumer(natsConsumer) // Health check should succeed - yield* subscriber.healthCheck + yield* consumer.healthCheck }) ).pipe(Effect.provide(testJetStream), TestServices.provideLive)) }) diff --git a/packages/nats/test/NATSSubscriber.test.ts b/packages/nats/test/NATSConsumer.test.ts similarity index 72% rename from packages/nats/test/NATSSubscriber.test.ts rename to packages/nats/test/NATSConsumer.test.ts index 3b06410..e209ee5 100644 --- a/packages/nats/test/NATSSubscriber.test.ts +++ b/packages/nats/test/NATSConsumer.test.ts @@ -1,24 +1,24 @@ import type { Mock } from "@effect/vitest" import { describe, expect, it, vi } from "@effect/vitest" -import { Effect, TestServices } from "effect" +import { Effect, Layer, TestServices } from "effect" +import * as NATSConsumer from "../src/NATSConsumer.js" import * as NATSMessage from "../src/NATSMessage.js" -import * as NATSPublisher from "../src/NATSPublisher.js" -import * as NATSSubscriber from "../src/NATSSubscriber.js" +import * as NATSProducer from "../src/NATSProducer.js" import { testConnection } from "./dependencies.js" // Use unique subject for this test file to avoid conflicts -const TEST_SUBJECT = "nats.subscriber.test.subject" +const TEST_SUBJECT = "nats.consumer.test.subject" const publishAndAssertConsume = ( - { content, onMessage, publisher, times }: { - publisher: NATSPublisher.NATSPublisher + { content, onMessage, producer, times }: { + producer: NATSProducer.NATSProducer onMessage: Mock<(message: NATSMessage.NATSMessage) => void> content: Uint8Array times: number } ) => Effect.gen(function*() { - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: content }) @@ -32,30 +32,30 @@ const publishAndAssertConsume = ( })) }) -describe("NATSSubscriber", { sequential: true }, () => { - describe("subscribe", () => { +describe("NATSConsumer", { sequential: true }, () => { + describe("serve", () => { it.effect("Should consume published events", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() - // IMPORTANT: For NATS Core, subscriber MUST be started BEFORE publishing + // IMPORTANT: For NATS Core, consumer MUST be started BEFORE publishing // because there is no persistence - messages are fire-and-forget - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT) + const consumer = yield* NATSConsumer.make(TEST_SUBJECT) const onMessage = vi.fn<(message: NATSMessage.NATSMessage) => void>() - // Start the subscription - yield* Effect.fork(subscriber.subscribe(Effect.gen(function*() { + // Start the subscription using Layer.launch + yield* Effect.fork(Layer.launch(consumer.serve(Effect.gen(function*() { const message = yield* NATSMessage.NATSConsumeMessage onMessage(message) - }))) + })))) // Give the subscription time to start yield* Effect.sleep("100 millis") // Message 1 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 1"), times: 1 @@ -63,7 +63,7 @@ describe("NATSSubscriber", { sequential: true }, () => { // Message 2 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 2"), times: 2 @@ -71,21 +71,21 @@ describe("NATSSubscriber", { sequential: true }, () => { // Message 3 yield* publishAndAssertConsume({ - publisher, + producer, onMessage, content: new TextEncoder().encode("Message 3"), times: 3 }) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive)) + }).pipe(Effect.provide(testConnection), TestServices.provideLive)) it.effect("Should NOT receive messages published before subscription started (no persistence)", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() const onMessage = vi.fn<(message: NATSMessage.NATSMessage) => void>() // Publish BEFORE subscribing - this message will be lost - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message published before subscription") }) @@ -93,13 +93,13 @@ describe("NATSSubscriber", { sequential: true }, () => { // Wait a bit to ensure the message is sent yield* Effect.sleep("100 millis") - // Now start the subscriber - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT) + // Now start the consumer + const consumer = yield* NATSConsumer.make(TEST_SUBJECT) - yield* Effect.fork(subscriber.subscribe(Effect.gen(function*() { + yield* Effect.fork(Layer.launch(consumer.serve(Effect.gen(function*() { const message = yield* NATSMessage.NATSConsumeMessage onMessage(message) - }))) + })))) // Give the subscription time to start yield* Effect.sleep("100 millis") @@ -108,7 +108,7 @@ describe("NATSSubscriber", { sequential: true }, () => { expect(onMessage).toHaveBeenCalledTimes(0) // Now publish a message AFTER subscription - this should be received - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message published after subscription") }) @@ -120,15 +120,15 @@ describe("NATSSubscriber", { sequential: true }, () => { expect(onMessage).toHaveBeenCalledWith(expect.objectContaining({ subject: TEST_SUBJECT })) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive)) + }).pipe(Effect.provide(testConnection), TestServices.provideLive)) }) - describe("interruptable subscribers", { sequential: true }, () => { + describe("interruptable consumers", { sequential: true }, () => { it.effect( "Should interrupt the handler if the subscription fiber is interrupted", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() const onHandlingStarted = vi.fn<(message: NATSMessage.NATSMessage) => void>() const onHandlingFinished = vi.fn<(message: NATSMessage.NATSMessage) => void>() @@ -140,15 +140,15 @@ describe("NATSSubscriber", { sequential: true }, () => { onHandlingFinished(message) }) - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT) + const consumer = yield* NATSConsumer.make(TEST_SUBJECT) - // Start the subscription - const subscriptionFiber = yield* Effect.fork(subscriber.subscribe(handler)) + // Start the subscription using Layer.launch + const subscriptionFiber = yield* Effect.fork(Layer.launch(consumer.serve(handler))) // Give the subscription time to start yield* Effect.sleep("100 millis") - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will be interrupted") }) @@ -165,13 +165,13 @@ describe("NATSSubscriber", { sequential: true }, () => { // The message handling should be interrupted (not finished) expect(onHandlingFinished).toHaveBeenCalledTimes(0) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive), + }).pipe(Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 } ) - it.effect("Should not interrupt the handler if the subscriber is uninterruptible", () => + it.effect("Should not interrupt the handler if the consumer is uninterruptible", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() const onHandlingStarted = vi.fn<(message: NATSMessage.NATSMessage) => void>() const onHandlingFinished = vi.fn<(message: NATSMessage.NATSMessage) => void>() @@ -183,15 +183,15 @@ describe("NATSSubscriber", { sequential: true }, () => { onHandlingFinished(message) }) - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT, undefined, { uninterruptible: true }) + const consumer = yield* NATSConsumer.make(TEST_SUBJECT, undefined, { uninterruptible: true }) - // Start the subscription - const subscriptionFiber = yield* Effect.fork(subscriber.subscribe(handler)) + // Start the subscription using Layer.launch + const subscriptionFiber = yield* Effect.fork(Layer.launch(consumer.serve(handler))) // Give the subscription time to start yield* Effect.sleep("100 millis") - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will NOT be interrupted") }) @@ -206,13 +206,13 @@ describe("NATSSubscriber", { sequential: true }, () => { // The subscription should be uninterrupted - wait for the message to be consumed yield* Effect.sleep("300 millis") expect(onHandlingFinished).toHaveBeenCalledTimes(1) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 }) + }).pipe(Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 }) it.effect( "Should timeout the handler when handlerTimeout is set", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() const onHandlingStarted = vi.fn<(message: NATSMessage.NATSMessage) => void>() const onHandlingFinished = vi.fn<(message: NATSMessage.NATSMessage) => void>() @@ -225,17 +225,17 @@ describe("NATSSubscriber", { sequential: true }, () => { onHandlingFinished(message) }) - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT, undefined, { + const consumer = yield* NATSConsumer.make(TEST_SUBJECT, undefined, { handlerTimeout: "200 millis" }) - // Start the subscription - yield* Effect.fork(subscriber.subscribe(handler)) + // Start the subscription using Layer.launch + yield* Effect.fork(Layer.launch(consumer.serve(handler))) // Give the subscription time to start yield* Effect.sleep("100 millis") - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("My Message that will timeout") }) @@ -246,7 +246,7 @@ describe("NATSSubscriber", { sequential: true }, () => { // Handler started but did not finish due to timeout expect(onHandlingStarted).toHaveBeenCalledTimes(1) expect(onHandlingFinished).toHaveBeenCalledTimes(0) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive), + }).pipe(Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 } ) }) @@ -254,7 +254,7 @@ describe("NATSSubscriber", { sequential: true }, () => { describe("error handling", () => { it.effect("Should continue processing messages when handler fails", () => Effect.gen(function*() { - const publisher = yield* NATSPublisher.make() + const producer = yield* NATSProducer.make() const onHandlingStarted = vi.fn<(message: NATSMessage.NATSMessage) => void>() const onHandlingFinished = vi.fn<(message: NATSMessage.NATSMessage) => void>() @@ -273,16 +273,16 @@ describe("NATSSubscriber", { sequential: true }, () => { onHandlingFinished(message) }) - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT) + const consumer = yield* NATSConsumer.make(TEST_SUBJECT) - // Start the subscription - yield* Effect.fork(subscriber.subscribe(handler)) + // Start the subscription using Layer.launch + yield* Effect.fork(Layer.launch(consumer.serve(handler))) // Give the subscription time to start yield* Effect.sleep("100 millis") // First message - will fail - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message that will fail") }) @@ -292,7 +292,7 @@ describe("NATSSubscriber", { sequential: true }, () => { expect(onHandlingFinished).toHaveBeenCalledTimes(0) // Second message - should succeed - yield* publisher.publish({ + yield* producer.send({ subject: TEST_SUBJECT, payload: new TextEncoder().encode("Message that will succeed") }) @@ -300,16 +300,16 @@ describe("NATSSubscriber", { sequential: true }, () => { yield* Effect.sleep("200 millis") expect(onHandlingStarted).toHaveBeenCalledTimes(2) expect(onHandlingFinished).toHaveBeenCalledTimes(1) - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 }) + }).pipe(Effect.provide(testConnection), TestServices.provideLive), { timeout: 15000 }) }) describe("healthCheck", () => { it.effect("Should succeed when subscription is healthy", () => Effect.gen(function*() { - const subscriber = yield* NATSSubscriber.make(TEST_SUBJECT) + const consumer = yield* NATSConsumer.make(TEST_SUBJECT) // Health check should succeed - yield* subscriber.healthCheck - }).pipe(Effect.scoped, Effect.provide(testConnection), TestServices.provideLive)) + yield* consumer.healthCheck + }).pipe(Effect.provide(testConnection), TestServices.provideLive)) }) })