PubSub.SlidingStrategy — API

PubSub.SlidingStrategyclasseffect/PubSub.ts:2603

SlidingStrategy<A>

Represents the sliding strategy for bounded PubSub values.

When to use

Use to keep the most recent messages when the PubSub is at capacity.

Details

New messages are accepted by evicting older messages from the bounded PubSub.

Gotchas

Slow subscribers may miss older messages that are evicted before they are consumed.

Example (Applying a sliding strategy)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  // Create PubSub with sliding strategy
  const pubsub = yield* PubSub.sliding<string>(2)

  // Or explicitly create with sliding strategy
  const customPubsub = yield* PubSub.make<string>({
    atomicPubSub: () => PubSub.makeAtomicBounded(2),
    strategy: () => new PubSub.SlidingStrategy()
  })

  yield* Effect.scoped(Effect.gen(function*() {
    const subscription = yield* PubSub.subscribe(pubsub)

    // Publish messages that exceed capacity
    yield* PubSub.publish(pubsub, "msg1") // stored
    yield* PubSub.publish(pubsub, "msg2") // stored
    yield* PubSub.publish(pubsub, "msg3") // "msg1" evicted, "msg3" stored
    yield* PubSub.publish(pubsub, "msg4") // "msg2" evicted, "msg4" stored

    // Subscribers will see the most recent messages
    const messages = yield* PubSub.takeAll(subscription)
    console.log("Recent messages:", messages) // ["msg3", "msg4"]
  }))
})

models

Source effect/PubSub.ts:260356 lines

export class SlidingStrategy<in out A> implements PubSub.Strategy<A> {
  get shutdown(): Effect.Effect<void> {
    return Effect.void
  }

  handleSurplus(
    pubsub: PubSub.Atomic<A>,
    subscribers: PubSub.Subscribers<A>,
    elements: Iterable<A>,
    _isShutdown: MutableRef.MutableRef<boolean>
  ): Effect.Effect<boolean> {
    return Effect.sync(() => {
      this.slidingPublishUnsafe(pubsub, elements)
      this.completeSubscribersUnsafe(pubsub, subscribers)
      return true
    })
  }

  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 {
    return strategyCompletePollersUnsafe(this, pubsub, subscribers, subscription, pollers)
  }

  completeSubscribersUnsafe(pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>): void {
    return strategyCompleteSubscribersUnsafe(this, pubsub, subscribers)
  }

  slidingPublishUnsafe(pubsub: PubSub.Atomic<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
        }
      }
    }
  }
}

Referenced by 1 symbols

PubSub.sliding