Lorem Ipsum MCP Server

import * as Chunk from "../Chunk.js" import type * as Deferred from "../Deferred.js" import type * as Effect from "../Effect.js" import * as Effectable from "../Effectable.js" import { dual, pipe } from "../Function.js" import * as MutableQueue from "../MutableQueue.js" import * as MutableRef from "../MutableRef.js" import { nextPow2 } from "../Number.js" import * as Option from "../Option.js" import { pipeArguments } from "../Pipeable.js" import type * as PubSub from "../PubSub.js" import type * as Queue from "../Queue.js" import type * as Scope from "../Scope.js" import * as core from "./core.js" import * as executionStrategy from "./executionStrategy.js" import * as fiberRuntime from "./fiberRuntime.js" import * as queue from "./queue.js" const AbsentValue = Symbol.for("effect/PubSub/AbsentValue") type AbsentValue = typeof AbsentValue /** @internal */ export interface AtomicPubSub<in out A> { readonly capacity: number isEmpty(): boolean isFull(): boolean size(): number publish(value: A): boolean publishAll(elements: Iterable<A>): Chunk.Chunk<A> slide(): void subscribe(): Subscription<A> replayWindow(): ReplayWindow<A> } /** @internal */ interface Subscription<out A> { isEmpty(): boolean size(): number poll<D>(default_: D): A | D pollUpTo(n: number): Chunk.Chunk<A> unsubscribe(): void } /** @internal */ type Subscribers<A> = Map< Subscription<A>, Set<MutableQueue.MutableQueue<Deferred.Deferred<A>>> > const addSubscribers = <A>( subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ) => (subscribers: Subscribers<A>) => { if (!subscribers.has(subscription)) { subscribers.set(subscription, new Set()) } const set = subscribers.get(subscription)! set.add(pollers) } const removeSubscribers = <A>( subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ) => (subscribers: Subscribers<A>) => { if (!subscribers.has(subscription)) { return } const set = subscribers.get(subscription)! set.delete(pollers) if (set.size === 0) { subscribers.delete(subscription) } } /** @internal */ export const bounded = <A>( capacity: number | { readonly capacity: number readonly replay?: number | undefined } ): Effect.Effect<PubSub.PubSub<A>> => core.suspend(() => { const pubsub = makeBoundedPubSub<A>(capacity) return makePubSub(pubsub, new BackPressureStrategy()) }) /** @internal */ export const dropping = <A>( capacity: number | { readonly capacity: number readonly replay?: number | undefined } ): Effect.Effect<PubSub.PubSub<A>> => core.suspend(() => { const pubsub = makeBoundedPubSub<A>(capacity) return makePubSub(pubsub, new DroppingStrategy()) }) /** @internal */ export const sliding = <A>( capacity: number | { readonly capacity: number readonly replay?: number | undefined } ): Effect.Effect<PubSub.PubSub<A>> => core.suspend(() => { const pubsub = makeBoundedPubSub<A>(capacity) return makePubSub(pubsub, new SlidingStrategy()) }) /** @internal */ export const unbounded = <A>(options?: { readonly replay?: number | undefined }): Effect.Effect<PubSub.PubSub<A>> => core.suspend(() => { const pubsub = makeUnboundedPubSub<A>(options) return makePubSub(pubsub, new DroppingStrategy()) }) /** @internal */ export const capacity = <A>(self: PubSub.PubSub<A>): number => self.capacity() /** @internal */ export const size = <A>(self: PubSub.PubSub<A>): Effect.Effect<number> => self.size /** @internal */ export const isFull = <A>(self: PubSub.PubSub<A>): Effect.Effect<boolean> => self.isFull /** @internal */ export const isEmpty = <A>(self: PubSub.PubSub<A>): Effect.Effect<boolean> => self.isEmpty /** @internal */ export const shutdown = <A>(self: PubSub.PubSub<A>): Effect.Effect<void> => self.shutdown /** @internal */ export const isShutdown = <A>(self: PubSub.PubSub<A>): Effect.Effect<boolean> => self.isShutdown /** @internal */ export const awaitShutdown = <A>(self: PubSub.PubSub<A>): Effect.Effect<void> => self.awaitShutdown /** @internal */ export const publish = dual< <A>(value: A) => (self: PubSub.PubSub<A>) => Effect.Effect<boolean>, <A>(self: PubSub.PubSub<A>, value: A) => Effect.Effect<boolean> >(2, (self, value) => self.publish(value)) /** @internal */ export const publishAll = dual< <A>(elements: Iterable<A>) => (self: PubSub.PubSub<A>) => Effect.Effect<boolean>, <A>(self: PubSub.PubSub<A>, elements: Iterable<A>) => Effect.Effect<boolean> >(2, (self, elements) => self.publishAll(elements)) /** @internal */ export const subscribe = <A>(self: PubSub.PubSub<A>): Effect.Effect<Queue.Dequeue<A>, never, Scope.Scope> => self.subscribe /** @internal */ const makeBoundedPubSub = <A>( capacity: number | { readonly capacity: number readonly replay?: number | undefined } ): AtomicPubSub<A> => { const options = typeof capacity === "number" ? { capacity } : capacity ensureCapacity(options.capacity) const replayBuffer = options.replay && options.replay > 0 ? new ReplayBuffer<A>(Math.ceil(options.replay)) : undefined if (options.capacity === 1) { return new BoundedPubSubSingle(replayBuffer) } else if (nextPow2(options.capacity) === options.capacity) { return new BoundedPubSubPow2(options.capacity, replayBuffer) } else { return new BoundedPubSubArb(options.capacity, replayBuffer) } } /** @internal */ const makeUnboundedPubSub = <A>(options?: { readonly replay?: number | undefined }): AtomicPubSub<A> => new UnboundedPubSub(options?.replay ? new ReplayBuffer(options.replay) : undefined) /** @internal */ const makeSubscription = <A>( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, strategy: PubSubStrategy<A> ): Effect.Effect<Queue.Dequeue<A>> => core.map(core.deferredMake<void>(), (deferred) => unsafeMakeSubscription( pubsub, subscribers, pubsub.subscribe(), MutableQueue.unbounded<Deferred.Deferred<A>>(), deferred, MutableRef.make(false), strategy )) /** @internal */ export const unsafeMakeSubscription = <A>( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>, shutdownHook: Deferred.Deferred<void>, shutdownFlag: MutableRef.MutableRef<boolean>, strategy: PubSubStrategy<A> ): Queue.Dequeue<A> => new SubscriptionImpl( pubsub, subscribers, subscription, pollers, shutdownHook, shutdownFlag, strategy, pubsub.replayWindow() ) /** @internal */ class BoundedPubSubArb<in out A> implements AtomicPubSub<A> { array: Array<A> publisherIndex = 0 subscribers: Array<number> subscriberCount = 0 subscribersIndex = 0 constructor(readonly capacity: number, readonly replayBuffer: ReplayBuffer<A> | undefined) { this.array = Array.from({ length: capacity }) this.subscribers = Array.from({ length: capacity }) } replayWindow(): ReplayWindow<A> { return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow } isEmpty(): boolean { return this.publisherIndex === this.subscribersIndex } isFull(): boolean { return this.publisherIndex === this.subscribersIndex + this.capacity } size(): number { return this.publisherIndex - this.subscribersIndex } publish(value: A): boolean { if (this.isFull()) { return false } if (this.subscriberCount !== 0) { const index = this.publisherIndex % this.capacity this.array[index] = value this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 } if (this.replayBuffer) { this.replayBuffer.offer(value) } return true } publishAll(elements: Iterable<A>): Chunk.Chunk<A> { if (this.subscriberCount === 0) { if (this.replayBuffer) { this.replayBuffer.offerAll(elements) } return Chunk.empty() } const chunk = Chunk.fromIterable(elements) const n = chunk.length const size = this.publisherIndex - this.subscribersIndex const available = this.capacity - size const forPubSub = Math.min(n, available) if (forPubSub === 0) { return chunk } let iteratorIndex = 0 const publishAllIndex = this.publisherIndex + forPubSub while (this.publisherIndex !== publishAllIndex) { const a = Chunk.unsafeGet(chunk, iteratorIndex++) const index = this.publisherIndex % this.capacity this.array[index] = a this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 if (this.replayBuffer) { this.replayBuffer.offer(a) } } return Chunk.drop(chunk, iteratorIndex) } slide(): void { if (this.subscribersIndex !== this.publisherIndex) { const index = this.subscribersIndex % this.capacity this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 } if (this.replayBuffer) { this.replayBuffer.slide() } } subscribe(): Subscription<A> { this.subscriberCount += 1 return new BoundedPubSubArbSubscription(this, this.publisherIndex, false) } } class BoundedPubSubArbSubscription<in out A> implements Subscription<A> { constructor( private self: BoundedPubSubArb<A>, private subscriberIndex: number, private unsubscribed: boolean ) { } isEmpty(): boolean { return ( this.unsubscribed || this.self.publisherIndex === this.subscriberIndex || this.self.publisherIndex === this.self.subscribersIndex ) } size() { if (this.unsubscribed) { return 0 } return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex) } poll<D>(default_: D): A | D { if (this.unsubscribed) { return default_ } this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) if (this.subscriberIndex !== this.self.publisherIndex) { const index = this.subscriberIndex % this.self.capacity const elem = this.self.array[index]! this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } this.subscriberIndex += 1 return elem } return default_ } pollUpTo(n: number): Chunk.Chunk<A> { if (this.unsubscribed) { return Chunk.empty() } this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) const size = this.self.publisherIndex - this.subscriberIndex const toPoll = Math.min(n, size) if (toPoll <= 0) { return Chunk.empty() } const builder: Array<A> = [] const pollUpToIndex = this.subscriberIndex + toPoll while (this.subscriberIndex !== pollUpToIndex) { const index = this.subscriberIndex % this.self.capacity const a = this.self.array[index] as A this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } builder.push(a) this.subscriberIndex += 1 } return Chunk.fromIterable(builder) } unsubscribe(): void { if (!this.unsubscribed) { this.unsubscribed = true this.self.subscriberCount -= 1 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) while (this.subscriberIndex !== this.self.publisherIndex) { const index = this.subscriberIndex % this.self.capacity this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } this.subscriberIndex += 1 } } } } /** @internal */ class BoundedPubSubPow2<in out A> implements AtomicPubSub<A> { array: Array<A> mask: number publisherIndex = 0 subscribers: Array<number> subscriberCount = 0 subscribersIndex = 0 constructor(readonly capacity: number, readonly replayBuffer: ReplayBuffer<A> | undefined) { this.array = Array.from({ length: capacity }) this.mask = capacity - 1 this.subscribers = Array.from({ length: capacity }) } replayWindow(): ReplayWindow<A> { return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow } isEmpty(): boolean { return this.publisherIndex === this.subscribersIndex } isFull(): boolean { return this.publisherIndex === this.subscribersIndex + this.capacity } size(): number { return this.publisherIndex - this.subscribersIndex } publish(value: A): boolean { if (this.isFull()) { return false } if (this.subscriberCount !== 0) { const index = this.publisherIndex & this.mask this.array[index] = value this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 } if (this.replayBuffer) { this.replayBuffer.offer(value) } return true } publishAll(elements: Iterable<A>): Chunk.Chunk<A> { if (this.subscriberCount === 0) { if (this.replayBuffer) { this.replayBuffer.offerAll(elements) } return Chunk.empty() } const chunk = Chunk.fromIterable(elements) const n = chunk.length const size = this.publisherIndex - this.subscribersIndex const available = this.capacity - size const forPubSub = Math.min(n, available) if (forPubSub === 0) { return chunk } let iteratorIndex = 0 const publishAllIndex = this.publisherIndex + forPubSub while (this.publisherIndex !== publishAllIndex) { const elem = Chunk.unsafeGet(chunk, iteratorIndex++) const index = this.publisherIndex & this.mask this.array[index] = elem this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 if (this.replayBuffer) { this.replayBuffer.offer(elem) } } return Chunk.drop(chunk, iteratorIndex) } slide(): void { if (this.subscribersIndex !== this.publisherIndex) { const index = this.subscribersIndex & this.mask this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 } if (this.replayBuffer) { this.replayBuffer.slide() } } subscribe(): Subscription<A> { this.subscriberCount += 1 return new BoundedPubSubPow2Subscription(this, this.publisherIndex, false) } } /** @internal */ class BoundedPubSubPow2Subscription<in out A> implements Subscription<A> { constructor( private self: BoundedPubSubPow2<A>, private subscriberIndex: number, private unsubscribed: boolean ) { } isEmpty(): boolean { return ( this.unsubscribed || this.self.publisherIndex === this.subscriberIndex || this.self.publisherIndex === this.self.subscribersIndex ) } size() { if (this.unsubscribed) { return 0 } return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex) } poll<D>(default_: D): A | D { if (this.unsubscribed) { return default_ } this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) if (this.subscriberIndex !== this.self.publisherIndex) { const index = this.subscriberIndex & this.self.mask const elem = this.self.array[index]! this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } this.subscriberIndex += 1 return elem } return default_ } pollUpTo(n: number): Chunk.Chunk<A> { if (this.unsubscribed) { return Chunk.empty() } this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) const size = this.self.publisherIndex - this.subscriberIndex const toPoll = Math.min(n, size) if (toPoll <= 0) { return Chunk.empty() } const builder: Array<A> = [] const pollUpToIndex = this.subscriberIndex + toPoll while (this.subscriberIndex !== pollUpToIndex) { const index = this.subscriberIndex & this.self.mask const elem = this.self.array[index] as A this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } builder.push(elem) this.subscriberIndex += 1 } return Chunk.fromIterable(builder) } unsubscribe(): void { if (!this.unsubscribed) { this.unsubscribed = true this.self.subscriberCount -= 1 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex) while (this.subscriberIndex !== this.self.publisherIndex) { const index = this.subscriberIndex & this.self.mask this.self.subscribers[index] -= 1 if (this.self.subscribers[index] === 0) { this.self.array[index] = AbsentValue as unknown as A this.self.subscribersIndex += 1 } this.subscriberIndex += 1 } } } } /** @internal */ class BoundedPubSubSingle<in out A> implements AtomicPubSub<A> { publisherIndex = 0 subscriberCount = 0 subscribers = 0 value: A = AbsentValue as unknown as A readonly capacity = 1 constructor(readonly replayBuffer: ReplayBuffer<A> | undefined) {} replayWindow(): ReplayWindow<A> { return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow } pipe() { return pipeArguments(this, arguments) } isEmpty(): boolean { return this.subscribers === 0 } isFull(): boolean { return !this.isEmpty() } size(): number { return this.isEmpty() ? 0 : 1 } publish(value: A): boolean { if (this.isFull()) { return false } if (this.subscriberCount !== 0) { this.value = value this.subscribers = this.subscriberCount this.publisherIndex += 1 } if (this.replayBuffer) { this.replayBuffer.offer(value) } return true } publishAll(elements: Iterable<A>): Chunk.Chunk<A> { if (this.subscriberCount === 0) { if (this.replayBuffer) { this.replayBuffer.offerAll(elements) } return Chunk.empty() } const chunk = Chunk.fromIterable(elements) if (Chunk.isEmpty(chunk)) { return chunk } if (this.publish(Chunk.unsafeHead(chunk))) { return Chunk.drop(chunk, 1) } else { return chunk } } slide(): void { if (this.isFull()) { this.subscribers = 0 this.value = AbsentValue as unknown as A } if (this.replayBuffer) { this.replayBuffer.slide() } } subscribe(): Subscription<A> { this.subscriberCount += 1 return new BoundedPubSubSingleSubscription(this, this.publisherIndex, false) } } /** @internal */ class BoundedPubSubSingleSubscription<in out A> implements Subscription<A> { constructor( private self: BoundedPubSubSingle<A>, private subscriberIndex: number, private unsubscribed: boolean ) { } isEmpty(): boolean { return ( this.unsubscribed || this.self.subscribers === 0 || this.subscriberIndex === this.self.publisherIndex ) } size() { return this.isEmpty() ? 0 : 1 } poll<D>(default_: D): A | D { if (this.isEmpty()) { return default_ } const elem = this.self.value this.self.subscribers -= 1 if (this.self.subscribers === 0) { this.self.value = AbsentValue as unknown as A } this.subscriberIndex += 1 return elem } pollUpTo(n: number): Chunk.Chunk<A> { if (this.isEmpty() || n < 1) { return Chunk.empty() } const a = this.self.value this.self.subscribers -= 1 if (this.self.subscribers === 0) { this.self.value = AbsentValue as unknown as A } this.subscriberIndex += 1 return Chunk.of(a) } unsubscribe(): void { if (!this.unsubscribed) { this.unsubscribed = true this.self.subscriberCount -= 1 if (this.subscriberIndex !== this.self.publisherIndex) { this.self.subscribers -= 1 if (this.self.subscribers === 0) { this.self.value = AbsentValue as unknown as A } } } } } /** @internal */ interface Node<out A> { value: A | AbsentValue subscribers: number next: Node<A> | null } /** @internal */ class UnboundedPubSub<in out A> implements AtomicPubSub<A> { publisherHead: Node<A> = { value: AbsentValue, subscribers: 0, next: null } publisherTail = this.publisherHead publisherIndex = 0 subscribersIndex = 0 readonly capacity = Number.MAX_SAFE_INTEGER constructor(readonly replayBuffer: ReplayBuffer<A> | undefined) {} replayWindow(): ReplayWindow<A> { return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow } isEmpty(): boolean { return this.publisherHead === this.publisherTail } isFull(): boolean { return false } size(): number { return this.publisherIndex - this.subscribersIndex } publish(value: A): boolean { const subscribers = this.publisherTail.subscribers if (subscribers !== 0) { this.publisherTail.next = { value, subscribers, next: null } this.publisherTail = this.publisherTail.next this.publisherIndex += 1 } if (this.replayBuffer) { this.replayBuffer.offer(value) } return true } publishAll(elements: Iterable<A>): Chunk.Chunk<A> { if (this.publisherTail.subscribers !== 0) { for (const a of elements) { this.publish(a) } } else if (this.replayBuffer) { this.replayBuffer.offerAll(elements) } return Chunk.empty() } slide(): void { if (this.publisherHead !== this.publisherTail) { this.publisherHead = this.publisherHead.next! this.publisherHead.value = AbsentValue this.subscribersIndex += 1 } if (this.replayBuffer) { this.replayBuffer.slide() } } subscribe(): Subscription<A> { this.publisherTail.subscribers += 1 return new UnboundedPubSubSubscription( this, this.publisherTail, this.publisherIndex, false ) } } /** @internal */ class UnboundedPubSubSubscription<in out A> implements Subscription<A> { constructor( private self: UnboundedPubSub<A>, private subscriberHead: Node<A>, private subscriberIndex: number, private unsubscribed: boolean ) { } isEmpty(): boolean { if (this.unsubscribed) { return true } let empty = true let loop = true while (loop) { if (this.subscriberHead === this.self.publisherTail) { loop = false } else { if (this.subscriberHead.next!.value !== AbsentValue) { empty = false loop = false } else { this.subscriberHead = this.subscriberHead.next! this.subscriberIndex += 1 } } } return empty } size() { if (this.unsubscribed) { return 0 } return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex) } poll<D>(default_: D): A | D { if (this.unsubscribed) { return default_ } let loop = true let polled: A | D = default_ while (loop) { if (this.subscriberHead === this.self.publisherTail) { loop = false } else { const elem = this.subscriberHead.next!.value if (elem !== AbsentValue) { polled = elem this.subscriberHead.subscribers -= 1 if (this.subscriberHead.subscribers === 0) { this.self.publisherHead = this.self.publisherHead.next! this.self.publisherHead.value = AbsentValue this.self.subscribersIndex += 1 } loop = false } this.subscriberHead = this.subscriberHead.next! this.subscriberIndex += 1 } } return polled } pollUpTo(n: number): Chunk.Chunk<A> { const builder: Array<A> = [] const default_ = AbsentValue let i = 0 while (i !== n) { const a = this.poll(default_ as unknown as A) if (a === default_) { i = n } else { builder.push(a) i += 1 } } return Chunk.fromIterable(builder) } unsubscribe(): void { if (!this.unsubscribed) { this.unsubscribed = true this.self.publisherTail.subscribers -= 1 while (this.subscriberHead !== this.self.publisherTail) { if (this.subscriberHead.next!.value !== AbsentValue) { this.subscriberHead.subscribers -= 1 if (this.subscriberHead.subscribers === 0) { this.self.publisherHead = this.self.publisherHead.next! this.self.publisherHead.value = AbsentValue this.self.subscribersIndex += 1 } } this.subscriberHead = this.subscriberHead.next! } } } } /** @internal */ class SubscriptionImpl<in out A> extends Effectable.Class<A> implements Queue.Dequeue<A> { [queue.DequeueTypeId] = queue.dequeueVariance constructor( readonly pubsub: AtomicPubSub<A>, readonly subscribers: Subscribers<A>, readonly subscription: Subscription<A>, readonly pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>, readonly shutdownHook: Deferred.Deferred<void>, readonly shutdownFlag: MutableRef.MutableRef<boolean>, readonly strategy: PubSubStrategy<A>, readonly replayWindow: ReplayWindow<A> ) { super() } commit() { return this.take } pipe() { return pipeArguments(this, arguments) } capacity(): number { return this.pubsub.capacity } isActive(): boolean { return !MutableRef.get(this.shutdownFlag) } get size(): Effect.Effect<number> { return core.suspend(() => MutableRef.get(this.shutdownFlag) ? core.interrupt : core.succeed(this.subscription.size() + this.replayWindow.remaining) ) } unsafeSize(): Option.Option<number> { if (MutableRef.get(this.shutdownFlag)) { return Option.none() } return Option.some(this.subscription.size() + this.replayWindow.remaining) } get isFull(): Effect.Effect<boolean> { return core.suspend(() => MutableRef.get(this.shutdownFlag) ? core.interrupt : core.succeed(this.subscription.size() === this.capacity()) ) } get isEmpty(): Effect.Effect<boolean> { return core.map(this.size, (size) => size === 0) } get shutdown(): Effect.Effect<void> { return core.uninterruptible( core.withFiberRuntime<void>((state) => { MutableRef.set(this.shutdownFlag, true) return pipe( fiberRuntime.forEachParUnbounded( unsafePollAllQueue(this.pollers), (d) => core.deferredInterruptWith(d, state.id()), false ), core.zipRight(core.sync(() => { this.subscribers.delete(this.subscription) this.subscription.unsubscribe() this.strategy.unsafeOnPubSubEmptySpace(this.pubsub, this.subscribers) })), core.whenEffect(core.deferredSucceed(this.shutdownHook, void 0)), core.asVoid ) }) ) } get isShutdown(): Effect.Effect<boolean> { return core.sync(() => MutableRef.get(this.shutdownFlag)) } get awaitShutdown(): Effect.Effect<void> { return core.deferredAwait(this.shutdownHook) } get take(): Effect.Effect<A> { return core.withFiberRuntime((state) => { if (MutableRef.get(this.shutdownFlag)) { return core.interrupt } if (this.replayWindow.remaining > 0) { const message = this.replayWindow.take()! return core.succeed(message) } const message = MutableQueue.isEmpty(this.pollers) ? this.subscription.poll(MutableQueue.EmptyMutableQueue) : MutableQueue.EmptyMutableQueue if (message === MutableQueue.EmptyMutableQueue) { const deferred = core.deferredUnsafeMake<A>(state.id()) return pipe( core.suspend(() => { pipe(this.pollers, MutableQueue.offer(deferred)) pipe(this.subscribers, addSubscribers(this.subscription, this.pollers)) this.strategy.unsafeCompletePollers( this.pubsub, this.subscribers, this.subscription, this.pollers ) return MutableRef.get(this.shutdownFlag) ? core.interrupt : core.deferredAwait(deferred) }), core.onInterrupt(() => core.sync(() => unsafeRemove(this.pollers, deferred))) ) } else { this.strategy.unsafeOnPubSubEmptySpace(this.pubsub, this.subscribers) return core.succeed(message) } }) } get takeAll(): Effect.Effect<Chunk.Chunk<A>> { return core.suspend(() => { if (MutableRef.get(this.shutdownFlag)) { return core.interrupt } const as = MutableQueue.isEmpty(this.pollers) ? unsafePollAllSubscription(this.subscription) : Chunk.empty() this.strategy.unsafeOnPubSubEmptySpace(this.pubsub, this.subscribers) if (this.replayWindow.remaining > 0) { return core.succeed(Chunk.appendAll(this.replayWindow.takeAll(), as)) } return core.succeed(as) }) } takeUpTo(this: this, max: number): Effect.Effect<Chunk.Chunk<A>> { return core.suspend(() => { if (MutableRef.get(this.shutdownFlag)) { return core.interrupt } let replay: Chunk.Chunk<A> | undefined = undefined if (this.replayWindow.remaining >= max) { const as = this.replayWindow.takeN(max) return core.succeed(as) } else if (this.replayWindow.remaining > 0) { replay = this.replayWindow.takeAll() max = max - replay.length } const as = MutableQueue.isEmpty(this.pollers) ? unsafePollN(this.subscription, max) : Chunk.empty() this.strategy.unsafeOnPubSubEmptySpace(this.pubsub, this.subscribers) return replay ? core.succeed(Chunk.appendAll(replay, as)) : core.succeed(as) }) } takeBetween(min: number, max: number): Effect.Effect<Chunk.Chunk<A>> { return core.suspend(() => takeRemainderLoop(this, min, max, Chunk.empty())) } } /** @internal */ const takeRemainderLoop = <A>( self: Queue.Dequeue<A>, min: number, max: number, acc: Chunk.Chunk<A> ): Effect.Effect<Chunk.Chunk<A>> => { if (max < min) { return core.succeed(acc) } return pipe( self.takeUpTo(max), core.flatMap((bs) => { const remaining = min - bs.length if (remaining === 1) { return pipe(self.take, core.map((b) => pipe(acc, Chunk.appendAll(bs), Chunk.append(b)))) } if (remaining > 1) { return pipe( self.take, core.flatMap((b) => takeRemainderLoop( self, remaining - 1, max - bs.length - 1, pipe(acc, Chunk.appendAll(bs), Chunk.append(b)) ) ) ) } return core.succeed(pipe(acc, Chunk.appendAll(bs))) }) ) } /** @internal */ class PubSubImpl<in out A> implements PubSub.PubSub<A> { readonly [queue.EnqueueTypeId] = queue.enqueueVariance readonly [queue.DequeueTypeId] = queue.dequeueVariance constructor( readonly pubsub: AtomicPubSub<A>, readonly subscribers: Subscribers<A>, readonly scope: Scope.Scope.Closeable, readonly shutdownHook: Deferred.Deferred<void>, readonly shutdownFlag: MutableRef.MutableRef<boolean>, readonly strategy: PubSubStrategy<A> ) {} capacity(): number { return this.pubsub.capacity } get size(): Effect.Effect<number> { return core.suspend(() => MutableRef.get(this.shutdownFlag) ? core.interrupt : core.sync(() => this.pubsub.size()) ) } unsafeSize(): Option.Option<number> { if (MutableRef.get(this.shutdownFlag)) { return Option.none() } return Option.some(this.pubsub.size()) } get isFull(): Effect.Effect<boolean> { return core.map(this.size, (size) => size === this.capacity()) } get isEmpty(): Effect.Effect<boolean> { return core.map(this.size, (size) => size === 0) } get awaitShutdown(): Effect.Effect<void> { return core.deferredAwait(this.shutdownHook) } get isShutdown(): Effect.Effect<boolean> { return core.sync(() => MutableRef.get(this.shutdownFlag)) } get shutdown(): Effect.Effect<void> { return core.uninterruptible(core.withFiberRuntime((state) => { pipe(this.shutdownFlag, MutableRef.set(true)) return pipe( this.scope.close(core.exitInterrupt(state.id())), core.zipRight(this.strategy.shutdown), core.whenEffect(core.deferredSucceed(this.shutdownHook, void 0)), core.asVoid ) })) } publish(value: A): Effect.Effect<boolean> { return core.suspend(() => { if (MutableRef.get(this.shutdownFlag)) { return core.interrupt } if (this.pubsub.publish(value)) { this.strategy.unsafeCompleteSubscribers(this.pubsub, this.subscribers) return core.succeed(true) } return this.strategy.handleSurplus( this.pubsub, this.subscribers, Chunk.of(value), this.shutdownFlag ) }) } isActive(): boolean { return !MutableRef.get(this.shutdownFlag) } unsafeOffer(value: A): boolean { if (MutableRef.get(this.shutdownFlag)) { return false } if ((this.pubsub as AtomicPubSub<unknown>).publish(value)) { this.strategy.unsafeCompleteSubscribers(this.pubsub, this.subscribers) return true } return false } publishAll(elements: Iterable<A>): Effect.Effect<boolean> { return core.suspend(() => { if (MutableRef.get(this.shutdownFlag)) { return core.interrupt } const surplus = unsafePublishAll(this.pubsub, elements) this.strategy.unsafeCompleteSubscribers(this.pubsub, this.subscribers) if (Chunk.isEmpty(surplus)) { return core.succeed(true) } return this.strategy.handleSurplus( this.pubsub, this.subscribers, surplus, this.shutdownFlag ) }) } get subscribe(): Effect.Effect<Queue.Dequeue<A>, never, Scope.Scope> { const acquire = core.tap( fiberRuntime.all([ this.scope.fork(executionStrategy.sequential), makeSubscription(this.pubsub, this.subscribers, this.strategy) ]), (tuple) => tuple[0].addFinalizer(() => tuple[1].shutdown) ) return core.map( fiberRuntime.acquireRelease(acquire, (tuple, exit) => tuple[0].close(exit)), (tuple) => tuple[1] ) } offer(value: A): Effect.Effect<boolean> { return this.publish(value) } offerAll(elements: Iterable<A>): Effect.Effect<boolean> { return this.publishAll(elements) } pipe() { return pipeArguments(this, arguments) } } /** @internal */ export const makePubSub = <A>( pubsub: AtomicPubSub<A>, strategy: PubSubStrategy<A> ): Effect.Effect<PubSub.PubSub<A>> => core.flatMap( fiberRuntime.scopeMake(), (scope) => core.map(core.deferredMake<void>(), (deferred) => unsafeMakePubSub( pubsub, new Map(), scope, deferred, MutableRef.make(false), strategy )) ) /** @internal */ export const unsafeMakePubSub = <A>( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, scope: Scope.Scope.Closeable, shutdownHook: Deferred.Deferred<void>, shutdownFlag: MutableRef.MutableRef<boolean>, strategy: PubSubStrategy<A> ): PubSub.PubSub<A> => new PubSubImpl(pubsub, subscribers, scope, shutdownHook, shutdownFlag, strategy) /** @internal */ const ensureCapacity = (capacity: number): void => { if (capacity <= 0) { throw new core.InvalidPubSubCapacityException(`Cannot construct PubSub with capacity of ${capacity}`) } } /** @internal */ const unsafeCompleteDeferred = <A>(deferred: Deferred.Deferred<A>, a: A): void => { core.deferredUnsafeDone(deferred, core.succeed(a)) } /** @internal */ const unsafeOfferAll = <A>(queue: MutableQueue.MutableQueue<A>, as: Iterable<A>): Chunk.Chunk<A> => { return pipe(queue, MutableQueue.offerAll(as)) } /** @internal */ const unsafePollAllQueue = <A>(queue: MutableQueue.MutableQueue<A>): Chunk.Chunk<A> => { return pipe(queue, MutableQueue.pollUpTo(Number.POSITIVE_INFINITY)) } /** @internal */ const unsafePollAllSubscription = <A>(subscription: Subscription<A>): Chunk.Chunk<A> => { return subscription.pollUpTo(Number.POSITIVE_INFINITY) } /** @internal */ const unsafePollN = <A>(subscription: Subscription<A>, max: number): Chunk.Chunk<A> => { return subscription.pollUpTo(max) } /** @internal */ const unsafePublishAll = <A>(pubsub: AtomicPubSub<A>, as: Iterable<A>): Chunk.Chunk<A> => { return pubsub.publishAll(as) } /** @internal */ const unsafeRemove = <A>(queue: MutableQueue.MutableQueue<A>, value: A): void => { unsafeOfferAll( queue, pipe(unsafePollAllQueue(queue), Chunk.filter((elem) => elem !== value)) ) } // ----------------------------------------------------------------------------- // PubSub.Strategy // ----------------------------------------------------------------------------- /** * A `PubSubStrategy<A>` describes the protocol for how publishers and subscribers * will communicate with each other through the `PubSub`. * * @internal */ export interface PubSubStrategy<in out A> { /** * Describes any finalization logic associated with this strategy. */ readonly shutdown: Effect.Effect<void> /** * Describes how publishers should signal to subscribers that they are * waiting for space to become available in the `PubSub`. */ handleSurplus( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef.MutableRef<boolean> ): Effect.Effect<boolean> /** * Describes how subscribers should signal to publishers waiting for space * to become available in the `PubSub` that space may be available. */ unsafeOnPubSubEmptySpace( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A> ): void /** * Describes 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. */ unsafeCompletePollers( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ): void /** * Describes how publishers should signal to subscribers waiting for * additional values from the `PubSub` that new values are available. */ unsafeCompleteSubscribers( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A> ): void } /** * A strategy that applies back pressure to publishers when the `PubSub` is at * capacity. This guarantees that all subscribers will receive all messages * published to the `PubSub` while they are subscribed. However, it creates the * risk that a slow subscriber will slow down the rate at which messages * are published and received by other subscribers. * * @internal */ class BackPressureStrategy<in out A> implements PubSubStrategy<A> { publishers: MutableQueue.MutableQueue< readonly [A, Deferred.Deferred<boolean>, boolean] > = MutableQueue.unbounded() get shutdown(): Effect.Effect<void> { return core.flatMap(core.fiberId, (fiberId) => core.flatMap( core.sync(() => unsafePollAllQueue(this.publishers)), (publishers) => fiberRuntime.forEachConcurrentDiscard( publishers, ([_, deferred, last]) => last ? pipe(core.deferredInterruptWith(deferred, fiberId), core.asVoid) : core.void, false, false ) )) } handleSurplus( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef.MutableRef<boolean> ): Effect.Effect<boolean> { return core.withFiberRuntime((state) => { const deferred = core.deferredUnsafeMake<boolean>(state.id()) return pipe( core.suspend(() => { this.unsafeOffer(elements, deferred) this.unsafeOnPubSubEmptySpace(pubsub, subscribers) this.unsafeCompleteSubscribers(pubsub, subscribers) return MutableRef.get(isShutdown) ? core.interrupt : core.deferredAwait(deferred) }), core.onInterrupt(() => core.sync(() => this.unsafeRemove(deferred))) ) }) } unsafeOnPubSubEmptySpace( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A> ): void { let keepPolling = true while (keepPolling && !pubsub.isFull()) { const publisher = pipe(this.publishers, MutableQueue.poll(MutableQueue.EmptyMutableQueue)) if (publisher === MutableQueue.EmptyMutableQueue) { keepPolling = false } else { const published = pubsub.publish(publisher[0]) if (published && publisher[2]) { unsafeCompleteDeferred(publisher[1], true) } else if (!published) { unsafeOfferAll( this.publishers, pipe(unsafePollAllQueue(this.publishers), Chunk.prepend(publisher)) ) } this.unsafeCompleteSubscribers(pubsub, subscribers) } } } unsafeCompletePollers( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ): void { return unsafeStrategyCompletePollers(this, pubsub, subscribers, subscription, pollers) } unsafeCompleteSubscribers(pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>): void { return unsafeStrategyCompleteSubscribers(this, pubsub, subscribers) } private unsafeOffer(elements: Iterable<A>, deferred: Deferred.Deferred<boolean>): void { const iterator = elements[Symbol.iterator]() let next: IteratorResult<A> = iterator.next() if (!next.done) { // eslint-disable-next-line no-constant-condition while (1) { const value = next.value next = iterator.next() if (next.done) { pipe( this.publishers, MutableQueue.offer([value, deferred, true as boolean] as const) ) break } pipe( this.publishers, MutableQueue.offer([value, deferred, false as boolean] as const) ) } } } unsafeRemove(deferred: Deferred.Deferred<boolean>): void { unsafeOfferAll( this.publishers, pipe(unsafePollAllQueue(this.publishers), Chunk.filter(([_, a]) => a !== deferred)) ) } } /** * A strategy that drops new messages when the `PubSub` is at capacity. This * guarantees that a slow subscriber will not slow down the rate at which * messages are published. However, it creates the risk that a slow * subscriber will slow down the rate at which messages are received by * other subscribers and that subscribers may not receive all messages * published to the `PubSub` while they are subscribed. * * @internal */ export class DroppingStrategy<in out A> implements PubSubStrategy<A> { get shutdown(): Effect.Effect<void> { return core.void } handleSurplus( _pubsub: AtomicPubSub<A>, _subscribers: Subscribers<A>, _elements: Iterable<A>, _isShutdown: MutableRef.MutableRef<boolean> ): Effect.Effect<boolean> { return core.succeed(false) } unsafeOnPubSubEmptySpace( _pubsub: AtomicPubSub<A>, _subscribers: Subscribers<A> ): void { // } unsafeCompletePollers( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ): void { return unsafeStrategyCompletePollers(this, pubsub, subscribers, subscription, pollers) } unsafeCompleteSubscribers(pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>): void { return unsafeStrategyCompleteSubscribers(this, pubsub, subscribers) } } /** * A strategy that adds new messages and drops old messages when the `PubSub` is * at capacity. This guarantees that a slow subscriber will not slow down * the rate at which messages are published and received by other * subscribers. However, it creates the risk that a slow subscriber will * not receive some messages published to the `PubSub` while it is subscribed. * * @internal */ export class SlidingStrategy<in out A> implements PubSubStrategy<A> { get shutdown(): Effect.Effect<void> { return core.void } handleSurplus( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, elements: Iterable<A>, _isShutdown: MutableRef.MutableRef<boolean> ): Effect.Effect<boolean> { return core.sync(() => { this.unsafeSlidingPublish(pubsub, elements) this.unsafeCompleteSubscribers(pubsub, subscribers) return true }) } unsafeOnPubSubEmptySpace( _pubsub: AtomicPubSub<A>, _subscribers: Subscribers<A> ): void { // } unsafeCompletePollers( pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ): void { return unsafeStrategyCompletePollers(this, pubsub, subscribers, subscription, pollers) } unsafeCompleteSubscribers(pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>): void { return unsafeStrategyCompleteSubscribers(this, pubsub, subscribers) } unsafeSlidingPublish(pubsub: AtomicPubSub<A>, elements: Iterable<A>): void { const it = elements[Symbol.iterator]() let next = it.next() if (!next.done && pubsub.capacity > 0) { let a = next.value let loop = true while (loop) { pubsub.slide() const pub = pubsub.publish(a) if (pub && (next = it.next()) && !next.done) { a = next.value } else if (pub) { loop = false } } } } } /** @internal */ const unsafeStrategyCompletePollers = <A>( strategy: PubSubStrategy<A>, pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>, subscription: Subscription<A>, pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>> ): void => { let keepPolling = true while (keepPolling && !subscription.isEmpty()) { const poller = pipe(pollers, MutableQueue.poll(MutableQueue.EmptyMutableQueue)) if (poller === MutableQueue.EmptyMutableQueue) { pipe(subscribers, removeSubscribers(subscription, pollers)) if (MutableQueue.isEmpty(pollers)) { keepPolling = false } else { pipe(subscribers, addSubscribers(subscription, pollers)) } } else { const pollResult = subscription.poll(MutableQueue.EmptyMutableQueue) if (pollResult === MutableQueue.EmptyMutableQueue) { unsafeOfferAll(pollers, pipe(unsafePollAllQueue(pollers), Chunk.prepend(poller))) } else { unsafeCompleteDeferred(poller, pollResult) strategy.unsafeOnPubSubEmptySpace(pubsub, subscribers) } } } } /** @internal */ const unsafeStrategyCompleteSubscribers = <A>( strategy: PubSubStrategy<A>, pubsub: AtomicPubSub<A>, subscribers: Subscribers<A> ): void => { for ( const [subscription, pollersSet] of subscribers ) { for (const pollers of pollersSet) { strategy.unsafeCompletePollers(pubsub, subscribers, subscription, pollers) } } } interface ReplayNode<A> { value: A | AbsentValue next: ReplayNode<A> | null } class ReplayBuffer<A> { constructor(readonly capacity: number) {} head: ReplayNode<A> = { value: AbsentValue, next: null } tail: ReplayNode<A> = this.head size = 0 index = 0 slide() { this.index++ } offer(a: A): void { this.tail.value = a this.tail.next = { value: AbsentValue, next: null } this.tail = this.tail.next if (this.size === this.capacity) { this.head = this.head.next! } else { this.size += 1 } } offerAll(as: Iterable<A>): void { for (const a of as) { this.offer(a) } } } interface ReplayWindow<A> { take(): A | undefined takeN(n: number): Chunk.Chunk<A> takeAll(): Chunk.Chunk<A> readonly remaining: number } class ReplayWindowImpl<A> implements ReplayWindow<A> { head: ReplayNode<A> index: number remaining: number constructor(readonly buffer: ReplayBuffer<A>) { this.index = buffer.index this.remaining = buffer.size this.head = buffer.head } fastForward() { while (this.index < this.buffer.index) { this.head = this.head.next! this.index++ } } take(): A | undefined { if (this.remaining === 0) { return undefined } else if (this.index < this.buffer.index) { this.fastForward() } this.remaining-- const value = this.head.value this.head = this.head.next! return value as A } takeN(n: number): Chunk.Chunk<A> { if (this.remaining === 0) { return Chunk.empty() } else if (this.index < this.buffer.index) { this.fastForward() } const len = Math.min(n, this.remaining) const items = new Array(len) for (let i = 0; i < len; i++) { const value = this.head.value as A this.head = this.head.next! items[i] = value } this.remaining -= len return Chunk.unsafeFromArray(items) } takeAll(): Chunk.Chunk<A> { return this.takeN(this.remaining) } } const emptyReplayWindow: ReplayWindow<never> = { remaining: 0, take: () => undefined, takeN: () => Chunk.empty(), takeAll: () => Chunk.empty() }