DurableQueue

7 exports Added in v1.0.0 Source

Constructors

make

Added in v1.0.0 Source

A DurableQueue wraps a PersistedQueue, providing a way to wait for items to finish processing using a DurableDeferred.

Signature

declare function make<Payload extends any, Success extends Any = Void, Error extends All = Never>(options: {
  readonly error?: Error;
  readonly idempotencyKey: (payload: Payload extends Fields ? View<Payload, "Type", TypeOptionalKeys<Payload>, TypeMutableKeys<Payload>> : Payload["Type"]) => string;
  readonly name: string;
  readonly payload: Payload;
  readonly success?: Success;
}): DurableQueue<Payload extends Fields ? Struct<Payload> : Payload, Success, Error>

Example

import { DurableQueue, Workflow } from "@effect/workflow"
import { Effect, Schema } from "effect"

// Define a DurableQueue that can be used to derive workers and offer items for
// processing.
const ApiQueue = DurableQueue.make({
  name: "ApiQueue",
  payload: {
    id: Schema.String
  },
  success: Schema.Void,
  error: Schema.Never,
  idempotencyKey(payload) {
    return payload.id
  }
})

const MyWorkflow = Workflow.make({
  name: "MyWorkflow",
  payload: {
    id: Schema.String
  },
  idempotencyKey: ({ id }) => id
})

const MyWorkflowLayer = MyWorkflow.toLayer(
  Effect.fn(function*() {
    // Add an item to the DurableQueue defined above.
    //
    // When the worker has finished processing the item, the workflow will
    // resume.
    //
    yield* DurableQueue.process(ApiQueue, { id: "api-call-1" })

    yield* Effect.log("Workflow succeeded!")
  })
)

// Define a worker layer that can process items from the DurableQueue.
const ApiWorker = DurableQueue.worker(
  ApiQueue,
  Effect.fn(function*({ id }) {
    yield* Effect.log(`Worker processing API call with id: ${id}`)
  }),
  { concurrency: 5 } // Process up to 5 items concurrently
)

Models

DurableQueue interface

Added in v1.0.0 Source

Signature

interface DurableQueue<Payload extends Schema.Schema.Any, Success extends Schema.Schema.Any = typeof Schema.Void, Error extends Schema.Schema.All = typeof Schema.Never> {
  readonly "~@effect/workflow/DurableQueue": "~@effect/workflow/DurableQueue";
  readonly deferred: DurableDeferred<Success, Error>;
  readonly idempotencyKey: (payload: Payload["Type"]) => string;
  readonly name: string;
  readonly payloadSchema: Payload;
}

Processing

process

Added in v1.0.0 Source

Signature

declare const process: <Payload extends Schema.Schema.Any, Success extends Schema.Schema.Any, Error extends Schema.Schema.All>(self: DurableQueue<Payload, Success, Error>, payload: Payload["Type"], options?: {
  readonly retrySchedule?: Schedule.Schedule<any, PersistedQueue.PersistedQueueError>;
}) => Effect.Effect<Success["Type"], Error["Type"], WorkflowEngine.WorkflowEngine | WorkflowEngine.WorkflowInstance | PersistedQueue.PersistedQueueFactory | Success["Context"] | Error["Context"] | Payload["Context"]>

Type IDs

TypeId

Added in v1.0.0 Source

Signature

declare const TypeId: "~@effect/workflow/DurableQueue"

TypeId type

Added in v1.0.0 Source

Signature

type TypeId = "~@effect/workflow/DurableQueue"

Worker

makeWorker

Added in v1.0.0 Source

Signature

declare const makeWorker: <Payload extends Schema.Schema.Any, Success extends Schema.Schema.Any, Error extends Schema.Schema.All, R>(self: DurableQueue<Payload, Success, Error>, f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>, options?: {
  readonly concurrency?: number;
}) => Effect.Effect<never, never, WorkflowEngine.WorkflowEngine | PersistedQueue.PersistedQueueFactory | R | Payload["Context"] | Success["Context"] | Error["Context"]>

worker

Added in v1.0.0 Source

Signature

declare const worker: <Payload extends Schema.Schema.Any, Success extends Schema.Schema.Any, Error extends Schema.Schema.All, R>(self: DurableQueue<Payload, Success, Error>, f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>, options?: {
  readonly concurrency?: number;
}) => Layer.Layer<never, never, WorkflowEngine.WorkflowEngine | PersistedQueue.PersistedQueueFactory | R | Payload["Context"] | Success["Context"] | Error["Context"]>