initiate redis-stream implementation

docker-compose.yaml

@@ -18,5 +18,15 @@ services:
       timeout: 5s
       retries: 5
 
+  redis:
+    image: redis:7-alpine
+    ports:
+      - 6379:6379
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
 volumes:
   rabbitmq_data:

packages/redis-stream/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 effect-messaging
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

packages/redis-stream/README.md

@@ -0,0 +1,64 @@
+# @effect-messaging/redis-stream
+
+A Redis Stream toolkit for Effect.
+
+## Installation
+
+```bash
+pnpm add @effect-messaging/redis-stream
+```
+
+## Usage
+
+### Publisher
+
+```typescript
+import {
+  RedisConnection,
+  RedisStreamPublisher
+} from "@effect-messaging/redis-stream"
+import { Effect } from "effect"
+
+const program = Effect.gen(function* () {
+  const publisher = yield* RedisStreamPublisher.make()
+
+  yield* publisher.publish({
+    stream: "my-stream",
+    data: { message: "Hello World" }
+  })
+})
+
+const runnable = program.pipe(
+  Effect.provide(RedisStreamPublisher.layer()),
+  Effect.provide(RedisConnection.layer({ host: "localhost", port: 6379 }))
+)
+
+Effect.runPromise(runnable)
+```
+
+### Subscriber
+
+```typescript
+import {
+  RedisConnection,
+  RedisStreamSubscriber
+} from "@effect-messaging/redis-stream"
+import { Effect } from "effect"
+
+const messageHandler = Effect.gen(function* () {
+  const message = yield* RedisStreamMessage.RedisStreamMessage
+  yield* Effect.logInfo(`Received: ${JSON.stringify(message.data)}`)
+})
+
+const program = Effect.gen(function* () {
+  const subscriber = yield* RedisStreamSubscriber.make("my-stream")
+  yield* subscriber.subscribe(messageHandler)
+})
+
+const runnable = program.pipe(
+  Effect.provide(RedisStreamSubscriber.layer("my-stream")),
+  Effect.provide(RedisConnection.layer({ host: "localhost", port: 6379 }))
+)
+
+Effect.runPromise(runnable)
+```

packages/redis-stream/docgen.json

@@ -0,0 +1,7 @@
+{
+    "src": "src",
+    "out": "docs",
+    "theme": "minimal",
+    "includeVersion": true,
+    "categorizeByGroup": true
+}

packages/redis-stream/examples/establish-connection.ts

@@ -0,0 +1,34 @@
+import { RedisConnection } from "@effect-messaging/redis-stream"
+import { Effect } from "effect"
+
+const program = Effect.gen(function*(_) {
+  const connection = yield* RedisConnection.RedisConnection
+
+  // Test the connection with a ping
+  const pong = yield* connection.ping
+  yield* Effect.logInfo(`Redis connection established: ${pong}`)
+
+  // Test basic Redis operations
+  const client = yield* connection.client
+  yield* Effect.tryPromise({
+    try: () => client.set("test-key", "test-value"),
+    catch: (error) => new Error(`Failed to set key: ${error}`)
+  })
+
+  const value = yield* Effect.tryPromise({
+    try: () => client.get("test-key"),
+    catch: (error) => new Error(`Failed to get key: ${error}`)
+  })
+
+  yield* Effect.logInfo(`Retrieved value: ${value}`)
+})
+
+const runnable = program.pipe(
+  Effect.provide(RedisConnection.layer({
+    host: "localhost",
+    port: 6379
+  }))
+)
+
+// Run the program
+Effect.runPromise(runnable)

packages/redis-stream/examples/publisher.ts

@@ -0,0 +1,31 @@
+import { RedisConnection, RedisStreamPublisher } from "@effect-messaging/redis-stream"
+import { Context, Effect } from "effect"
+
+class MyPublisher extends Context.Tag("MyPublisher")<MyPublisher, RedisStreamPublisher.RedisStreamPublisher>() {}
+
+const program = Effect.gen(function*(_) {
+  const publisher = yield* MyPublisher
+
+  yield* publisher.publish({
+    stream: "my-stream",
+    data: {
+      message: "Hello World",
+      timestamp: Date.now().toString(),
+      type: "greeting"
+    }
+  })
+
+  yield* Effect.logInfo("Message published successfully")
+})
+
+const runnable = program.pipe(
+  Effect.provideServiceEffect(MyPublisher, RedisStreamPublisher.make()),
+  // provide the Redis Connection dependency
+  Effect.provide(RedisConnection.layer({
+    host: "localhost",
+    port: 6379
+  }))
+)
+
+// Run the program
+Effect.runPromise(runnable)

packages/redis-stream/examples/subscriber.ts

@@ -0,0 +1,32 @@
+import { RedisConnection, RedisStreamMessage, RedisStreamSubscriber } from "@effect-messaging/redis-stream"
+import { Effect } from "effect"
+
+const messageHandler = Effect.gen(function*(_) {
+  const message = yield* RedisStreamMessage.RedisStreamMessage
+
+  // You can add your message processing logic here
+  yield* Effect.logInfo(`Received message: ${JSON.stringify(message.data)}`)
+  yield* Effect.logInfo(`Message ID: ${message.id}, Timestamp: ${message.timestamp}`)
+})
+
+const program = Effect.gen(function*(_) {
+  const subscriber = yield* RedisStreamSubscriber.make("my-stream", {
+    blockTimeout: 1000,
+    count: 10
+  })
+
+  // The subscriber will automatically handle message acknowledgment
+  // based on the success or failure of the message handler
+  yield* subscriber.subscribe(messageHandler)
+})
+
+const runnable = program.pipe(
+  // provide the Redis Connection dependency
+  Effect.provide(RedisConnection.layer({
+    host: "localhost",
+    port: 6379
+  }))
+)
+
+// Run the program
+Effect.runPromise(runnable)

packages/redis-stream/package.json

@@ -0,0 +1,54 @@
+{
+    "name": "@effect-messaging/redis-stream",
+    "version": "0.1.0",
+    "type": "module",
+    "license": "MIT",
+    "description": "A Redis Stream toolkit for Effect",
+    "homepage": "https://github.com/spiko-tech/effect-messaging",
+    "repository": {
+        "type": "git",
+        "url": "https://github.com/spiko-tech/effect-messaging.git",
+        "directory": "packages/redis-stream"
+    },
+    "bugs": {
+        "url": "https://github.com/spiko-tech/effect-messaging/issues"
+    },
+    "tags": [
+        "typescript",
+        "redis",
+        "streams"
+    ],
+    "keywords": [
+        "typescript",
+        "redis",
+        "streams"
+    ],
+    "publishConfig": {
+        "access": "public",
+        "directory": "dist",
+        "provenance": true
+    },
+    "scripts": {
+        "codegen": "build-utils prepare-v2",
+        "build": "pnpm build-esm && pnpm build-annotate && pnpm build-cjs && build-utils pack-v2",
+        "build-esm": "tsc -b tsconfig.build.json",
+        "build-cjs": "babel build/esm --plugins @babel/transform-export-namespace-from --plugins @babel/transform-modules-commonjs --out-dir build/cjs --source-maps",
+        "build-annotate": "babel build/esm --plugins annotate-pure-calls --out-dir build/esm --source-maps",
+        "check": "tsc -b tsconfig.json",
+        "test": "vitest",
+        "coverage": "vitest --coverage"
+    },
+    "devDependencies": {
+        "@effect-messaging/core": "workspace:^",
+        "@effect/platform": "^0.92.1",
+        "@types/node": "^22.14.1",
+        "effect": "^3.18.3",
+        "redis": "^4.6.12"
+    },
+    "peerDependencies": {
+        "@effect-messaging/core": "workspace:^",
+        "@effect/platform": "^0.92.1",
+        "effect": "^3.18.3",
+        "redis": "^4.6.12"
+    }
+}

packages/redis-stream/src/RedisConnection.ts

@@ -0,0 +1,108 @@
+/**
+ * @since 0.1.0
+ */
+import * as Context from "effect/Context"
+import type * as Duration from "effect/Duration"
+import * as Effect from "effect/Effect"
+import * as Layer from "effect/Layer"
+import type * as Schedule from "effect/Schedule"
+import type * as Scope from "effect/Scope"
+import type { RedisClientType } from "redis"
+import * as internal from "./internal/RedisConnection.js"
+import * as RedisError from "./RedisError.js"
+
+/**
+ * @category type ids
+ * @since 0.1.0
+ */
+export const TypeId: unique symbol = Symbol.for("@effect-messaging/redis-stream/RedisConnection")
+
+/**
+ * @category type ids
+ * @since 0.1.0
+ */
+export type TypeId = typeof TypeId
+
+/**
+ * @category models
+ * @since 0.1.0
+ */
+export interface RedisConnection {
+  readonly [TypeId]: TypeId
+  readonly client: Effect.Effect<RedisClientType, RedisError.RedisConnectionError, never>
+  readonly ping: Effect.Effect<string, RedisError.RedisConnectionError, never>
+  readonly quit: Effect.Effect<void, RedisError.RedisConnectionError, never>
+
+  /** @internal */
+  readonly close: (options: internal.CloseConnectionOptions) => Effect.Effect<void, never, never>
+}
+
+/**
+ * @category tags
+ * @since 0.1.0
+ */
+export const RedisConnection = Context.GenericTag<RedisConnection>("@effect-messaging/redis-stream/RedisConnection")
+
+/**
+ * @category models
+ * @since 0.1.0
+ */
+export type RedisConnectionOptions = {
+  host?: string
+  port?: number
+  password?: string
+  db?: number
+  retryConnectionSchedule?: Schedule.Schedule<unknown, RedisError.RedisConnectionError>
+  waitConnectionTimeout?: Duration.DurationInput
+}
+
+/**
+ * @category constructors
+ * @since 0.1.0
+ */
+export const make = (
+  options: RedisConnectionOptions = {}
+): Effect.Effect<RedisConnection, RedisError.RedisConnectionError, Scope.Scope> =>
+  Effect.gen(function*() {
+    const internalConnection = yield* internal.InternalRedisConnection
+    const provideInternal = Effect.provideService(internal.InternalRedisConnection, internalConnection)
+
+    const connection = yield* Effect.acquireRelease(
+      Effect.gen(function*() {
+        yield* internal.initiateConnection
+        return {
+          [TypeId]: TypeId as TypeId,
+          client: internal.getOrWaitClient.pipe(provideInternal),
+          ping: Effect.gen(function*() {
+            const client = yield* internal.getOrWaitClient.pipe(provideInternal)
+            return yield* Effect.tryPromise({
+              try: () => client.ping(),
+              catch: (error) => new RedisError.RedisConnectionError({ reason: "Ping failed", cause: error })
+            })
+          }) as Effect.Effect<string, RedisError.RedisConnectionError, never>,
+          quit: Effect.gen(function*() {
+            const client = yield* internal.getOrWaitClient.pipe(provideInternal)
+            return yield* Effect.tryPromise({
+              try: () => client.quit(),
+              catch: (error) => new RedisError.RedisConnectionError({ reason: "Quit failed", cause: error })
+            })
+          }) as Effect.Effect<void, RedisError.RedisConnectionError, never>,
+          close: (opts: internal.CloseConnectionOptions = {}) => internal.closeConnection(opts).pipe(provideInternal)
+        }
+      }),
+      (connection) => connection.close()
+    )
+    yield* Effect.forkScoped(internal.keepConnectionAlive)
+    yield* Effect.forkScoped(internal.monitorConnectionErrors)
+    return connection
+  }).pipe(
+    Effect.provideServiceEffect(internal.InternalRedisConnection, internal.InternalRedisConnection.new(options))
+  )
+
+/**
+ * @since 0.1.0
+ * @category Layers
+ */
+export const layer = (
+  options: RedisConnectionOptions = {}
+): Layer.Layer<RedisConnection, RedisError.RedisConnectionError> => Layer.scoped(RedisConnection, make(options))