@beignet/provider-broadcast-memory

Ephemeral BroadcastPort provider for local development, tests, and applications whose publishers and subscribers share one process. Use the Redis provider when workers or replicas run separately.

Install

bun add @beignet/core @beignet/provider-broadcast-memory zod
Copy

Beignet requires Node.js 22.12 or newer. Bun is also supported. The example below uses Zod for channel schemas.

Provider setup

import { createMemoryBroadcastProvider } from "@beignet/provider-broadcast-memory";

export const providers = [createMemoryBroadcastProvider()] as const;
Copy

Add this provider to your existing registry. Declare broadcast: BroadcastPort in AppPorts using the type from @beignet/core/broadcasting/server, and defer broadcast in infra/port-wiring.ts. The optional name setting changes the provider's registry name. No worker loop or environment variables are needed. Stopping the provider closes its active subscriptions.

beignet make broadcast issues.changes creates the channel, authorization stub, registry, endpoint, and this provider when the broadcast port is missing.

Direct use and tests

import { defineChannel } from "@beignet/core/broadcasting";
import { createMemoryBroadcast } from "@beignet/provider-broadcast-memory";
import { z } from "zod";

const changes = defineChannel("issues.changes", {
  params: z.object({ workspaceId: z.string() }),
  events: { changed: z.object({ issueId: z.string() }) },
});
const broadcast = createMemoryBroadcast();
const received = Promise.withResolvers<void>();
const subscription = broadcast.subscribe(changes, {
  params: { workspaceId: "workspace-1" },
  onEvent: event => {
    console.log(event.data.issueId);
    received.resolve();
  },
  onDisconnect: () => console.log("Reconnect and refetch"),
});
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
  await subscription.ready;
  await broadcast.publish(changes, {
    params: { workspaceId: "workspace-1" },
    event: "changed",
    data: { issueId: "issue-1" },
  });
  await Promise.race([
    received.promise,
    new Promise<never>((_, reject) => {
      timeout = setTimeout(() => reject(new Error("No hint received")), 5_000);
    }),
  ]);
} finally {
  clearTimeout(timeout);
  await subscription.unsubscribe();
}
Copy

This example prints issue-1 before unsubscribing. It waits up to five seconds for the listener because publish() itself does not wait for event callbacks. In an application, keep the subscription active for the component or process that owns it, then unsubscribe during that owner's cleanup.

createMemoryBroadcast({ instrumentation }) accepts a Beignet instrumentation target. Records use the broadcast watcher and omit payloads, parameters, and origin IDs. Each instance has isolated subscriptions. Receivers own their event data; changing it cannot change another receiver's data.

Publication succeeds with no listeners. It is not an acknowledgement from a browser. Schemas are validated on both sides; invalid envelopes or overflowing receiver queues interrupt the subscription. Await readiness before relying on live delivery, and always await unsubscribe() during cleanup.

Browser authorization belongs in explicit channel bindings exposed through createBroadcastRoute from @beignet/web or @beignet/next. The provider is a server-side primitive and does not authorize direct application publications.

See the broadcasting guide for browser subscriptions, React Query reconciliation, origin exclusion, notification delivery, and transaction/outbox ordering.