effect-pubsub-event-bus

PubSub Event Bus with Effect v4

Overview

Effect's PubSub module provides a typed, composable publish/subscribe primitive. Combined with Stream.fromPubSub and Effect.forkScoped, it replaces imperative callback-based event buses with a fully typed, stream-based subscription model where cleanup is automatic.

When to use this skill:

• Building intra-service communication (event bus, message broker)
• Replacing callback-based event systems with typed streams
• Implementing reactive patterns where services subscribe to domain events
• Managing per-instance event channels with scoped cleanup

Import Pattern

import { Effect, PubSub, Stream } from 'effect';

Core Pattern: Typed Event Bus Service

Define events as a discriminated union, then build a Bus service that publishes and subscribes using PubSub:

import { Effect, Layer, PubSub, Schema, Context, Stream } from 'effect';

// ─── Event Definitions ──────────────────────────────────────

class FileChanged extends Schema.TaggedClass<FileChanged>()('FileChanged', {
	path: Schema.String,
	kind: Schema.Literals(['created', 'modified', 'deleted'])
}) {}

class ConfigReloaded extends Schema.TaggedClass<ConfigReloaded>()(
	'ConfigReloaded',
	{
		source: Schema.String
	}
) {}

type BusEvent = FileChanged | ConfigReloaded;

// ─── Bus Service ─────────────────────────────────────────────

export namespace Bus {
	export interface Interface {
		readonly publish: (event: BusEvent) => Effect.Effect<void>;
		readonly subscribe: <T extends BusEvent>(
			eventClass: new (...args: ReadonlyArray<never>) => T
		) => Stream.Stream<T>;
		readonly subscribeAll: Stream.Stream<BusEvent>;
	}

	export class Service extends Context.Service<Service, Interface>()(
		'@app/Bus'
	) {}

	export const layer = Layer.effect(
		Service,
		Effect.gen(function* () {
			const pubsub = yield* PubSub.unbounded<BusEvent>();

			// Cleanup: shutdown PubSub when scope closes
			yield* Effect.addFinalizer(() => PubSub.shutdown(pubsub));

			const publish = Effect.fn('Bus.publish')(function* (
				event: BusEvent
			) {
				yield* PubSub.publish(pubsub, event);
			});

			const subscribe = <T extends BusEvent>(
				eventClass: new (...args: ReadonlyArray<never>) => T
			): Stream.Stream<T> =>
				Stream.fromPubSub(pubsub).pipe(
					Stream.filter((evt): evt is T => evt instanceof eventClass)
				);

			const subscribeAll = Stream.fromPubSub(pubsub);

			return Service.of({ publish, subscribe, subscribeAll });
		})
	);
}