<A>(value: A): (self: PubSub<A>) => boolean
<A>(self: PubSub<A>, value: A): booleanAttempts to publish a message synchronously without applying the PubSub strategy's effectful surplus handling.
When to use
Use when you need a non-blocking synchronous publish attempt where false
is an acceptable result when the message cannot be accepted immediately.
Details
Returns false if the PubSub is shut down or the message cannot be
accepted immediately, for example when a bounded PubSub is full. Prefer
publish when backpressure or sliding behavior should be honored.
Example (Publishing without suspending)
import { PubSub } from "effect"
declare const pubsub: PubSub.PubSub<string>
// Unsafe synchronous publish (non-blocking)
const published = PubSub.publishUnsafe(pubsub, "Hello")
if (published) {
console.log("Message published successfully")
} else {
console.log("Message dropped (PubSub full or shutdown)")
}
// Useful for scenarios where you don't want to suspend
const messages = ["msg1", "msg2", "msg3"]
const publishedCount =
messages.filter((msg) => PubSub.publishUnsafe(pubsub, msg)).length
console.log(`Published ${publishedCount} out of ${messages.length} messages`)publishingpublish
Source effect/PubSub.ts:96911 lines
export const const publishUnsafe: {
<A>(value: A): (self: PubSub<A>) => boolean
<A>(self: PubSub<A>, value: A): boolean
}
Attempts to publish a message synchronously without applying the PubSub
strategy's effectful surplus handling.
When to use
Use when you need a non-blocking synchronous publish attempt where false
is an acceptable result when the message cannot be accepted immediately.
Details
Returns false if the PubSub is shut down or the message cannot be
accepted immediately, for example when a bounded PubSub is full. Prefer
publish when backpressure or sliding behavior should be honored.
Example (Publishing without suspending)
import { PubSub } from "effect"
declare const pubsub: PubSub.PubSub<string>
// Unsafe synchronous publish (non-blocking)
const published = PubSub.publishUnsafe(pubsub, "Hello")
if (published) {
console.log("Message published successfully")
} else {
console.log("Message dropped (PubSub full or shutdown)")
}
// Useful for scenarios where you don't want to suspend
const messages = ["msg1", "msg2", "msg3"]
const publishedCount =
messages.filter((msg) => PubSub.publishUnsafe(pubsub, msg)).length
console.log(`Published ${publishedCount} out of ${messages.length} messages`)
publishUnsafe: {
<function (type parameter) A in <A>(value: A): (self: PubSub<A>) => booleanA>(value: Avalue: function (type parameter) A in <A>(value: A): (self: PubSub<A>) => booleanA): (self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface PubSub<in out A>A PubSub<A> is an asynchronous message hub into which publishers can publish
messages of type A and subscribers can subscribe to take messages of type
A.
Example (Publishing and subscribing to messages)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create a bounded PubSub with capacity 10
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe and consume messages
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages
yield* PubSub.publish(pubsub, "Hello")
yield* PubSub.publish(pubsub, "World")
const message1 = yield* PubSub.take(subscription)
const message2 = yield* PubSub.take(subscription)
console.log(message1, message2) // "Hello", "World"
}))
})
Companion namespace containing the low-level building blocks used by
PubSub, including atomic implementations, backing subscriptions, replay
windows, and delivery strategies.
PubSub<function (type parameter) A in <A>(value: A): (self: PubSub<A>) => booleanA>) => boolean
<function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA>(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface PubSub<in out A>A PubSub<A> is an asynchronous message hub into which publishers can publish
messages of type A and subscribers can subscribe to take messages of type
A.
Example (Publishing and subscribing to messages)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create a bounded PubSub with capacity 10
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe and consume messages
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages
yield* PubSub.publish(pubsub, "Hello")
yield* PubSub.publish(pubsub, "World")
const message1 = yield* PubSub.take(subscription)
const message2 = yield* PubSub.take(subscription)
console.log(message1, message2) // "Hello", "World"
}))
})
Companion namespace containing the low-level building blocks used by
PubSub, including atomic implementations, backing subscriptions, replay
windows, and delivery strategies.
PubSub<function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA>, value: Avalue: function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA): boolean
} = import dualdual(2, <function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA>(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface PubSub<in out A>A PubSub<A> is an asynchronous message hub into which publishers can publish
messages of type A and subscribers can subscribe to take messages of type
A.
Example (Publishing and subscribing to messages)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create a bounded PubSub with capacity 10
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe and consume messages
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages
yield* PubSub.publish(pubsub, "Hello")
yield* PubSub.publish(pubsub, "World")
const message1 = yield* PubSub.take(subscription)
const message2 = yield* PubSub.take(subscription)
console.log(message1, message2) // "Hello", "World"
}))
})
Companion namespace containing the low-level building blocks used by
PubSub, including atomic implementations, backing subscriptions, replay
windows, and delivery strategies.
PubSub<function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA>, value: Avalue: function (type parameter) A in <A>(self: PubSub<A>, value: A): booleanA): boolean => {
if (self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
PubSub<in out A>.shutdownFlag: MutableRef.MutableRef<boolean>(property) PubSub<in out A>.shutdownFlag: {
current: T;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
PubSub<A>.pubsub: PubSub.Atomic<A>(property) PubSub<A>.pubsub: {
capacity: number;
isEmpty: () => boolean;
isFull: () => boolean;
size: () => number;
publish: (value: A) => boolean;
publishAll: (elements: Iterable<A>) => Array<A>;
slide: () => void;
subscribe: () => PubSub.BackingSubscription<A>;
replayWindow: () => PubSub.ReplayWindow<A>;
}
PubSub<in out A>.Atomic<A>.publish(value: A): booleanpublish(value: Avalue)) {
self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
PubSub<A>.strategy: PubSub.Strategy<A>(property) PubSub<A>.strategy: {
shutdown: Effect.Effect<void>;
handleSurplus: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef.MutableRef<boolean>) => Effect.Effect<boolean>;
onPubSubEmptySpaceUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => void;
completePollersUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>, subscription: PubSub.BackingSubscription<A>, pollers: MutableList.MutableList<Deferred.Deferred<A, never>>) => void;
completeSubscribersUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => void;
}
PubSub<in out A>.Strategy<A>.completeSubscribersUnsafe(pubsub: PubSub<in out A>.Atomic<A>, subscribers: PubSub.Subscribers<A>): voidDescribes how publishers should signal to subscribers waiting for
additional values from the PubSub that new values are available.
completeSubscribersUnsafe(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
PubSub<A>.pubsub: PubSub.Atomic<A>(property) PubSub<A>.pubsub: {
capacity: number;
isEmpty: () => boolean;
isFull: () => boolean;
size: () => number;
publish: (value: A) => boolean;
publishAll: (elements: Iterable<A>) => Array<A>;
slide: () => void;
subscribe: () => PubSub.BackingSubscription<A>;
replayWindow: () => PubSub.ReplayWindow<A>;
}
self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
PubSub<A>.subscribers: PubSub.Subscribers<A>(property) PubSub<A>.subscribers: {
clear: () => void;
delete: (key: PubSub.BackingSubscription<A>) => boolean;
forEach: (callbackfn: (value: Set<MutableList.MutableList<Deferred.Deferred<A, never>>>, key: PubSub.BackingSubscription<A>, map: Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A, never>>>>) => void, thisArg?: any)…;
get: (key: PubSub.BackingSubscription<A>) => Set<MutableList.MutableList<Deferred.Deferred<A, never>>> | undefined;
has: (key: PubSub.BackingSubscription<A>) => boolean;
set: (key: PubSub.BackingSubscription<A>, value: Set<MutableList.MutableList<Deferred.Deferred<A, never>>>) => PubSub.Subscribers<A>;
size: number;
entries: () => MapIterator<[PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A, never>>>]>;
keys: () => MapIterator<PubSub.BackingSubscription<A>>;
values: () => MapIterator<Set<MutableList.MutableList<Deferred.Deferred<A, never>>>>;
}
Referenced by 1 symbols