DroppingStrategy<A>Represents the dropping strategy for bounded PubSub values.
When to use
Use to keep publishers fast by dropping new messages when the PubSub is at
capacity.
Details
A publish that arrives while the PubSub is full is dropped instead of
waiting for capacity.
Gotchas
Subscribers may miss messages published while they are subscribed.
Example (Applying a dropping strategy)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create PubSub with dropping strategy
const pubsub = yield* PubSub.dropping<string>(2)
// Or explicitly create with dropping strategy
const customPubsub = yield* PubSub.make<string>({
atomicPubSub: () => PubSub.makeAtomicBounded(2),
strategy: () => new PubSub.DroppingStrategy()
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill the PubSub
const pub1 = yield* PubSub.publish(pubsub, "msg1") // true
const pub2 = yield* PubSub.publish(pubsub, "msg2") // true
const pub3 = yield* PubSub.publish(pubsub, "msg3") // false (dropped)
console.log("Publication results:", [pub1, pub2, pub3]) // [true, true, false]
// Subscribers will only see the first two messages
const messages = yield* PubSub.takeAll(subscription)
console.log("Received messages:", messages) // ["msg1", "msg2"]
}))
})models
Source effect/PubSub.ts:251734 lines
export class class DroppingStrategy<in out A>class DroppingStrategy {
shutdown: Effect.Effect<void, never, never>;
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>>) => void;
completeSubscribersUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => void;
}
Represents the dropping strategy for bounded PubSub values.
When to use
Use to keep publishers fast by dropping new messages when the PubSub is at
capacity.
Details
A publish that arrives while the PubSub is full is dropped instead of
waiting for capacity.
Gotchas
Subscribers may miss messages published while they are subscribed.
Example (Applying a dropping strategy)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create PubSub with dropping strategy
const pubsub = yield* PubSub.dropping<string>(2)
// Or explicitly create with dropping strategy
const customPubsub = yield* PubSub.make<string>({
atomicPubSub: () => PubSub.makeAtomicBounded(2),
strategy: () => new PubSub.DroppingStrategy()
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill the PubSub
const pub1 = yield* PubSub.publish(pubsub, "msg1") // true
const pub2 = yield* PubSub.publish(pubsub, "msg2") // true
const pub3 = yield* PubSub.publish(pubsub, "msg3") // false (dropped)
console.log("Publication results:", [pub1, pub2, pub3]) // [true, true, false]
// Subscribers will only see the first two messages
const messages = yield* PubSub.takeAll(subscription)
console.log("Received messages:", messages) // ["msg1", "msg2"]
}))
})
DroppingStrategy<in out function (type parameter) A in DroppingStrategy<in out A>A> implements PubSub.interface PubSub<in out A>.Strategy<in out A>Strategy interface defining how PubSub handles backpressure and message distribution.
Strategy<function (type parameter) A in DroppingStrategy<in out A>A> {
get DroppingStrategy<in out A>.shutdown: Effect.Effect<void, never, never>(getter) DroppingStrategy<in out A>.shutdown: {
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;
}
Describes any finalization logic associated with this strategy.
shutdown(): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<void> {
return import EffectEffect.void
}
function DroppingStrategy(_pubsub: PubSub.Atomic<A>, _subscribers: PubSub.Subscribers<A>, _elements: Iterable<A>, _isShutdown: MutableRef.MutableRef<boolean>): Effect.Effect<boolean>Describes how publishers should signal to subscribers that they are
waiting for space to become available in the PubSub.
handleSurplus(
_pubsub: PubSub.Atomic<A>(parameter) _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>;
}
interface PubSub<in out A>.Atomic<in out A>Low-level atomic PubSub interface that handles the core message storage and retrieval.
Atomic<function (type parameter) A in DroppingStrategy<in out A>A>,
_subscribers: PubSub.Subscribers<A>(parameter) _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>>>>;
}
type PubSub<in out A>.Subscribers<A> = Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A>>>>Tracks the pollers currently waiting on each backing subscription.
Details
This type is part of the low-level PubSub.Strategy contract. Most
application code should use subscribe, take, and the other PubSub
operations instead of manipulating subscriber maps directly.
Subscribers<function (type parameter) A in DroppingStrategy<in out A>A>,
_elements: Iterable<A>_elements: interface Iterable<T, TReturn = any, TNext = any>Iterable<function (type parameter) A in DroppingStrategy<in out A>A>,
_isShutdown: MutableRef.MutableRef<boolean>(parameter) _isShutdown: {
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;
}
import MutableRefMutableRef.type MutableRef.MutableRef = /*unresolved*/ anyMutableRef<boolean>
): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<boolean> {
return import EffectEffect.succeed(false)
}
DroppingStrategy<in out A>.onPubSubEmptySpaceUnsafe(_pubsub: PubSub.Atomic<A>, _subscribers: PubSub.Subscribers<A>): voidDescribes how subscribers should signal to publishers waiting for space
to become available in the PubSub that space may be available.
onPubSubEmptySpaceUnsafe(
_pubsub: PubSub.Atomic<A>(parameter) _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>;
}
interface PubSub<in out A>.Atomic<in out A>Low-level atomic PubSub interface that handles the core message storage and retrieval.
Atomic<function (type parameter) A in DroppingStrategy<in out A>A>,
_subscribers: PubSub.Subscribers<A>(parameter) _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>>>>;
}
type PubSub<in out A>.Subscribers<A> = Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A>>>>Tracks the pollers currently waiting on each backing subscription.
Details
This type is part of the low-level PubSub.Strategy contract. Most
application code should use subscribe, take, and the other PubSub
operations instead of manipulating subscriber maps directly.
Subscribers<function (type parameter) A in DroppingStrategy<in out A>A>
): void {
//
}
function DroppingStrategy(pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>, subscription: PubSub.BackingSubscription<A>, pollers: MutableList.MutableList<Deferred.Deferred<A>>): voidDescribes how subscribers waiting for additional values from the PubSub
should take those values and signal to publishers that they are no
longer waiting for additional values.
completePollersUnsafe(
pubsub: PubSub.Atomic<A>(parameter) 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>;
}
interface PubSub<in out A>.Atomic<in out A>Low-level atomic PubSub interface that handles the core message storage and retrieval.
Atomic<function (type parameter) A in DroppingStrategy<in out A>A>,
subscribers: PubSub.Subscribers<A>(parameter) 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>>>>;
}
type PubSub<in out A>.Subscribers<A> = Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A>>>>Tracks the pollers currently waiting on each backing subscription.
Details
This type is part of the low-level PubSub.Strategy contract. Most
application code should use subscribe, take, and the other PubSub
operations instead of manipulating subscriber maps directly.
Subscribers<function (type parameter) A in DroppingStrategy<in out A>A>,
subscription: PubSub.BackingSubscription<A>(parameter) subscription: {
isEmpty: () => boolean;
size: () => number;
poll: () => typeof MutableList.Empty | A;
pollUpTo: (n: number) => Array<A>;
unsubscribe: () => void;
}
interface PubSub<in out A>.BackingSubscription<out A>Low-level subscription interface that handles message polling for individual subscribers.
BackingSubscription<function (type parameter) A in DroppingStrategy<in out A>A>,
pollers: MutableList.MutableList<
Deferred.Deferred<A>
>
(parameter) pollers: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
import MutableListMutableList.type MutableList.MutableList = /*unresolved*/ anyMutableList<import DeferredDeferred.type Deferred.Deferred = /*unresolved*/ anyDeferred<function (type parameter) A in DroppingStrategy<in out A>A>>
): void {
return const strategyCompletePollersUnsafe: <A>(
strategy: PubSub.Strategy<A>,
pubsub: PubSub.Atomic<A>,
subscribers: PubSub.Subscribers<A>,
subscription: PubSub.BackingSubscription<A>,
pollers: MutableList.MutableList<
Deferred.Deferred<A>
>
) => void
pubsub: PubSub.Atomic<A>(parameter) 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>;
}
subscribers: PubSub.Subscribers<A>(parameter) 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>>>>;
}
subscription: PubSub.BackingSubscription<A>(parameter) subscription: {
isEmpty: () => boolean;
size: () => number;
poll: () => typeof MutableList.Empty | A;
pollUpTo: (n: number) => Array<A>;
unsubscribe: () => void;
}
pollers: MutableList.MutableList<
Deferred.Deferred<A>
>
(parameter) pollers: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
DroppingStrategy<in out A>.completeSubscribersUnsafe(pubsub: PubSub.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(pubsub: PubSub.Atomic<A>(parameter) 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>;
}
interface PubSub<in out A>.Atomic<in out A>Low-level atomic PubSub interface that handles the core message storage and retrieval.
Atomic<function (type parameter) A in DroppingStrategy<in out A>A>, subscribers: PubSub.Subscribers<A>(parameter) 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>>>>;
}
type PubSub<in out A>.Subscribers<A> = Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A>>>>Tracks the pollers currently waiting on each backing subscription.
Details
This type is part of the low-level PubSub.Strategy contract. Most
application code should use subscribe, take, and the other PubSub
operations instead of manipulating subscriber maps directly.
Subscribers<function (type parameter) A in DroppingStrategy<in out A>A>): void {
return const strategyCompleteSubscribersUnsafe: <A>(strategy: PubSub<in out A>.Strategy<A>, pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => voidstrategyCompleteSubscribersUnsafe(this, pubsub: PubSub.Atomic<A>(parameter) 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>;
}
subscribers: PubSub.Subscribers<A>(parameter) 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 2 symbols