add PubSub replay option (#3135)

Co-authored-by: Tim Smart <tim.smart@arisechurch.com>

Author: tim-smart

.changeset/seven-ghosts-move.md

@@ -0,0 +1,18 @@
+---
+"effect": minor
+---
+
+add `replay` option to PubSub constructors
+
+This option adds a replay buffer in front of the given PubSub. The buffer will
+replay the last `n` messages to any new subscriber.
+
+```ts
+Effect.gen(function*() {
+  const messages = [1, 2, 3, 4, 5]
+  const pubsub = yield* PubSub.bounded<number>({ capacity: 16, replay: 3 })
+  yield* PubSub.publishAll(pubsub, messages)
+  const sub = yield* PubSub.subscribe(pubsub)
+  assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [3, 4, 5])
+}))
+```

packages/effect/src/PubSub.ts

@@ -46,7 +46,9 @@ export interface PubSub<in out A> extends Queue.Enqueue<A>, Pipeable {
  * @since 2.0.0
  * @category constructors
  */
-export const bounded: <A>(requestedCapacity: number) => Effect.Effect<PubSub<A>> = internal.bounded
+export const bounded: <A>(
+  capacity: number | { readonly capacity: number; readonly replay?: number | undefined }
+) => Effect.Effect<PubSub<A>> = internal.bounded
 
 /**
  * Creates a bounded `PubSub` with the dropping strategy. The `PubSub` will drop new
@@ -57,7 +59,9 @@ export const bounded: <A>(requestedCapacity: number) => Effect.Effect<PubSub<A>>
  * @since 2.0.0
  * @category constructors
  */
-export const dropping: <A>(requestedCapacity: number) => Effect.Effect<PubSub<A>> = internal.dropping
+export const dropping: <A>(
+  capacity: number | { readonly capacity: number; readonly replay?: number | undefined }
+) => Effect.Effect<PubSub<A>> = internal.dropping
 
 /**
  * Creates a bounded `PubSub` with the sliding strategy. The `PubSub` will add new
@@ -68,15 +72,18 @@ export const dropping: <A>(requestedCapacity: number) => Effect.Effect<PubSub<A>
  * @since 2.0.0
  * @category constructors
  */
-export const sliding: <A>(requestedCapacity: number) => Effect.Effect<PubSub<A>> = internal.sliding
+export const sliding: <A>(
+  capacity: number | { readonly capacity: number; readonly replay?: number | undefined }
+) => Effect.Effect<PubSub<A>> = internal.sliding
 
 /**
  * Creates an unbounded `PubSub`.
  *
  * @since 2.0.0
  * @category constructors
  */
-export const unbounded: <A>() => Effect.Effect<PubSub<A>> = internal.unbounded
+export const unbounded: <A>(options?: { readonly replay?: number | undefined }) => Effect.Effect<PubSub<A>> =
+  internal.unbounded
 
 /**
  *  Returns the number of elements the queue can hold.

packages/effect/src/internal/pubsub.ts

@@ -28,6 +28,7 @@ export interface AtomicPubSub<in out A> {
   publishAll(elements: Iterable<A>): Chunk.Chunk<A>
   slide(): void
   subscribe(): Subscription<A>
+  replayWindow(): ReplayWindow<A>
 }
 
 /** @internal */
@@ -73,32 +74,49 @@ const removeSubscribers = <A>(
 }
 
 /** @internal */
-export const bounded = <A>(requestedCapacity: number): Effect.Effect<PubSub.PubSub<A>> =>
-  pipe(
-    core.sync(() => makeBoundedPubSub<A>(requestedCapacity)),
-    core.flatMap((atomicPubSub) => makePubSub(atomicPubSub, new BackPressureStrategy()))
-  )
+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>(requestedCapacity: number): Effect.Effect<PubSub.PubSub<A>> =>
-  pipe(
-    core.sync(() => makeBoundedPubSub<A>(requestedCapacity)),
-    core.flatMap((atomicPubSub) => makePubSub(atomicPubSub, new DroppingStrategy()))
-  )
+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>(requestedCapacity: number): Effect.Effect<PubSub.PubSub<A>> =>
-  pipe(
-    core.sync(() => makeBoundedPubSub<A>(requestedCapacity)),
-    core.flatMap((atomicPubSub) => makePubSub(atomicPubSub, new SlidingStrategy()))
-  )
+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>(): Effect.Effect<PubSub.PubSub<A>> =>
-  pipe(
-    core.sync(() => makeUnboundedPubSub<A>()),
-    core.flatMap((atomicPubSub) => makePubSub(atomicPubSub, new DroppingStrategy()))
-  )
+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()
@@ -138,21 +156,28 @@ export const subscribe = <A>(self: PubSub.PubSub<A>): Effect.Effect<Queue.Dequeu
   self.subscribe
 
 /** @internal */
-const makeBoundedPubSub = <A>(requestedCapacity: number): AtomicPubSub<A> => {
-  ensureCapacity(requestedCapacity)
-  if (requestedCapacity === 1) {
-    return new BoundedPubSubSingle()
-  } else if (nextPow2(requestedCapacity) === requestedCapacity) {
-    return new BoundedPubSubPow2(requestedCapacity)
+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(requestedCapacity)
+    return new BoundedPubSubArb(options.capacity, replayBuffer)
   }
 }
 
 /** @internal */
-const makeUnboundedPubSub = <A>(): AtomicPubSub<A> => {
-  return new UnboundedPubSub()
-}
+const makeUnboundedPubSub = <A>(options?: {
+  readonly replay?: number | undefined
+}): AtomicPubSub<A> => new UnboundedPubSub(options?.replay ? new ReplayBuffer(options.replay) : undefined)
 
 /** @internal */
 const makeSubscription = <A>(
@@ -180,17 +205,17 @@ export const unsafeMakeSubscription = <A>(
   shutdownHook: Deferred.Deferred<void>,
   shutdownFlag: MutableRef.MutableRef<boolean>,
   strategy: PubSubStrategy<A>
-): Queue.Dequeue<A> => {
-  return new SubscriptionImpl(
+): Queue.Dequeue<A> =>
+  new SubscriptionImpl(
     pubsub,
     subscribers,
     subscription,
     pollers,
     shutdownHook,
     shutdownFlag,
-    strategy
+    strategy,
+    pubsub.replayWindow()
   )
-}
 
 /** @internal */
 class BoundedPubSubArb<in out A> implements AtomicPubSub<A> {
@@ -200,12 +225,13 @@ class BoundedPubSubArb<in out A> implements AtomicPubSub<A> {
   subscriberCount = 0
   subscribersIndex = 0
 
-  readonly capacity: number
+  constructor(readonly capacity: number, readonly replayBuffer: ReplayBuffer<A> | undefined) {
+    this.array = Array.from({ length: capacity })
+    this.subscribers = Array.from({ length: capacity })
+  }
 
-  constructor(requestedCapacity: number) {
-    this.array = Array.from({ length: requestedCapacity })
-    this.subscribers = Array.from({ length: requestedCapacity })
-    this.capacity = requestedCapacity
+  replayWindow(): ReplayWindow<A> {
+    return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
   }
 
   isEmpty(): boolean {
@@ -230,11 +256,17 @@ class BoundedPubSubArb<in out A> implements AtomicPubSub<A> {
       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)
@@ -253,6 +285,9 @@ class BoundedPubSubArb<in out A> implements AtomicPubSub<A> {
       this.array[index] = a
       this.subscribers[index] = this.subscriberCount
       this.publisherIndex += 1
+      if (this.replayBuffer) {
+        this.replayBuffer.offer(a)
+      }
     }
     return Chunk.drop(chunk, iteratorIndex)
   }
@@ -264,6 +299,9 @@ class BoundedPubSubArb<in out A> implements AtomicPubSub<A> {
       this.subscribers[index] = 0
       this.subscribersIndex += 1
     }
+    if (this.replayBuffer) {
+      this.replayBuffer.slide()
+    }
   }
 
   subscribe(): Subscription<A> {
@@ -368,13 +406,14 @@ class BoundedPubSubPow2<in out A> implements AtomicPubSub<A> {
   subscriberCount = 0
   subscribersIndex = 0
 
-  readonly capacity: number
+  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 })
+  }
 
-  constructor(requestedCapacity: number) {
-    this.array = Array.from({ length: requestedCapacity })
-    this.mask = requestedCapacity - 1
-    this.subscribers = Array.from({ length: requestedCapacity })
-    this.capacity = requestedCapacity
+  replayWindow(): ReplayWindow<A> {
+    return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
   }
 
   isEmpty(): boolean {
@@ -399,11 +438,17 @@ class BoundedPubSubPow2<in out A> implements AtomicPubSub<A> {
       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)
@@ -422,6 +467,9 @@ class BoundedPubSubPow2<in out A> implements AtomicPubSub<A> {
       this.array[index] = elem
       this.subscribers[index] = this.subscriberCount
       this.publisherIndex += 1
+      if (this.replayBuffer) {
+        this.replayBuffer.offer(elem)
+      }
     }
     return Chunk.drop(chunk, iteratorIndex)
   }
@@ -433,6 +481,9 @@ class BoundedPubSubPow2<in out A> implements AtomicPubSub<A> {
       this.subscribers[index] = 0
       this.subscribersIndex += 1
     }
+    if (this.replayBuffer) {
+      this.replayBuffer.slide()
+    }
   }
 
   subscribe(): Subscription<A> {
@@ -536,6 +587,11 @@ class BoundedPubSubSingle<in out A> implements AtomicPubSub<A> {
   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)
@@ -562,11 +618,17 @@ class BoundedPubSubSingle<in out A> implements AtomicPubSub<A> {
       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)
@@ -585,6 +647,9 @@ class BoundedPubSubSingle<in out A> implements AtomicPubSub<A> {
       this.subscribers = 0
       this.value = AbsentValue as unknown as A
     }
+    if (this.replayBuffer) {
+      this.replayBuffer.slide()
+    }
   }
 
   subscribe(): Subscription<A> {
@@ -673,6 +738,11 @@ class UnboundedPubSub<in out A> implements AtomicPubSub<A> {
   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
@@ -697,6 +767,9 @@ class UnboundedPubSub<in out A> implements AtomicPubSub<A> {
       this.publisherTail = this.publisherTail.next
       this.publisherIndex += 1
     }
+    if (this.replayBuffer) {
+      this.replayBuffer.offer(value)
+    }
     return true
   }
 
@@ -705,6 +778,8 @@ class UnboundedPubSub<in out A> implements AtomicPubSub<A> {
       for (const a of elements) {
         this.publish(a)
       }
+    } else if (this.replayBuffer) {
+      this.replayBuffer.offerAll(elements)
     }
     return Chunk.empty()
   }
@@ -715,6 +790,9 @@ class UnboundedPubSub<in out A> implements AtomicPubSub<A> {
       this.publisherHead.value = AbsentValue
       this.subscribersIndex += 1
     }
+    if (this.replayBuffer) {
+      this.replayBuffer.slide()
+    }
   }
 
   subscribe(): Subscription<A> {
@@ -841,9 +919,9 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
     readonly pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>,
     readonly shutdownHook: Deferred.Deferred<void>,
     readonly shutdownFlag: MutableRef.MutableRef<boolean>,
-    readonly strategy: PubSubStrategy<A>
-  ) {
-  }
+    readonly strategy: PubSubStrategy<A>,
+    readonly replayWindow: ReplayWindow<A>
+  ) {}
 
   pipe() {
     return pipeArguments(this, arguments)
@@ -861,19 +939,23 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
     return core.suspend(() =>
       MutableRef.get(this.shutdownFlag)
         ? core.interrupt
-        : core.succeed(this.subscription.size())
+        : 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())
+    return Option.some(this.subscription.size() + this.replayWindow.remaining)
   }
 
   get isFull(): Effect.Effect<boolean> {
-    return core.map(this.size, (size) => size === this.capacity())
+    return core.suspend(() =>
+      MutableRef.get(this.shutdownFlag)
+        ? core.interrupt
+        : core.succeed(this.subscription.size() === this.capacity())
+    )
   }
 
   get isEmpty(): Effect.Effect<boolean> {
@@ -915,6 +997,10 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
       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
@@ -950,6 +1036,9 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
         ? 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)
     })
   }
@@ -959,11 +1048,19 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
       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 core.succeed(as)
+      return replay ? core.succeed(Chunk.appendAll(replay, as)) : core.succeed(as)
     })
   }
 
@@ -1019,8 +1116,7 @@ class PubSubImpl<in out A> implements PubSub.PubSub<A> {
     readonly shutdownHook: Deferred.Deferred<void>,
     readonly shutdownFlag: MutableRef.MutableRef<boolean>,
     readonly strategy: PubSubStrategy<A>
-  ) {
-  }
+  ) {}
 
   capacity(): number {
     return this.pubsub.capacity
@@ -1075,7 +1171,7 @@ class PubSubImpl<in out A> implements PubSub.PubSub<A> {
         return core.interrupt
       }
 
-      if ((this.pubsub as AtomicPubSub<unknown>).publish(value)) {
+      if (this.pubsub.publish(value)) {
         this.strategy.unsafeCompleteSubscribers(this.pubsub, this.subscribers)
         return core.succeed(true)
       }
@@ -1179,9 +1275,7 @@ export const unsafeMakePubSub = <A>(
   shutdownHook: Deferred.Deferred<void>,
   shutdownFlag: MutableRef.MutableRef<boolean>,
   strategy: PubSubStrategy<A>
-): PubSub.PubSub<A> => {
-  return new PubSubImpl(pubsub, subscribers, scope, shutdownHook, shutdownFlag, strategy)
-}
+): PubSub.PubSub<A> => new PubSubImpl(pubsub, subscribers, scope, shutdownHook, shutdownFlag, strategy)
 
 /** @internal */
 const ensureCapacity = (capacity: number): void => {
@@ -1562,3 +1656,100 @@ const unsafeStrategyCompleteSubscribers = <A>(
     }
   }
 }
+
+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()
+}

packages/effect/test/PubSub.test.ts

@@ -1,4 +1,4 @@
-import { Option } from "effect"
+import { Chunk, Option } from "effect"
 import * as Array from "effect/Array"
 import * as Deferred from "effect/Deferred"
 import * as Effect from "effect/Effect"
@@ -661,4 +661,82 @@ describe("PubSub", () => {
       yield* PubSub.publishAll(pubsub, [1, 2])
       assert.deepStrictEqual(pubsub.unsafeSize(), Option.some(0))
     }))
+
+  describe("replay", () => {
+    it.scoped("unbounded", () =>
+      Effect.gen(function*() {
+        const messages = [1, 2, 3, 4, 5]
+        const pubsub = yield* PubSub.unbounded<number>({ replay: 3 })
+        yield* PubSub.publishAll(pubsub, messages)
+        const sub = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [3, 4, 5])
+      }))
+
+    it.effect("unbounded takeUpTo", () => {
+      const messages = [1, 2, 3, 4, 5]
+      return PubSub.unbounded<number>({ replay: 3 }).pipe(
+        Effect.flatMap((pubsub) =>
+          Effect.scoped(
+            Effect.gen(function*() {
+              yield* PubSub.publishAll(pubsub, messages)
+
+              const dequeue1 = yield* PubSub.subscribe(pubsub)
+              yield* PubSub.publish(pubsub, 6)
+              const dequeue2 = yield* PubSub.subscribe(pubsub)
+
+              assert.strictEqual(yield* Queue.size(dequeue1), 4)
+              assert.strictEqual(yield* Queue.size(dequeue2), 3)
+              assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeUpTo(dequeue1, 2)), [3, 4])
+              assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeUpTo(dequeue1, 2)), [5, 6])
+              assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeUpTo(dequeue2, 3)), [4, 5, 6])
+            })
+          )
+        )
+      )
+    })
+
+    it.scoped("dropping", () =>
+      Effect.gen(function*() {
+        const messages = [1, 2, 3, 4, 5]
+        const pubsub = yield* PubSub.dropping<number>({ capacity: 2, replay: 3 })
+
+        yield* PubSub.publishAll(pubsub, messages)
+        const sub = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [3, 4, 5])
+        yield* PubSub.publishAll(pubsub, [6, 7])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [6, 7])
+
+        const sub2 = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub2)), [5, 6, 7])
+
+        yield* PubSub.publishAll(pubsub, [8, 9, 10, 11])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [8, 9])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub2)), [8, 9])
+
+        const sub3 = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub3)), [7, 8, 9])
+      }))
+
+    it.scoped("sliding", () =>
+      Effect.gen(function*() {
+        const messages = [1, 2, 3, 4, 5]
+        const pubsub = yield* PubSub.sliding<number>({ capacity: 4, replay: 3 })
+
+        yield* PubSub.publishAll(pubsub, messages)
+        const sub = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(yield* Queue.take(sub), 3)
+        yield* PubSub.publishAll(pubsub, [6, 7, 8, 9, 10])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [5, 6, 7, 8, 9, 10])
+
+        const sub2 = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub2)), [8, 9, 10])
+
+        yield* PubSub.publishAll(pubsub, [11, 12, 13, 14, 15, 16])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub)), [13, 14, 15, 16])
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub2)), [13, 14, 15, 16])
+
+        const sub3 = yield* PubSub.subscribe(pubsub)
+        assert.deepStrictEqual(Chunk.toReadonlyArray(yield* Queue.takeAll(sub3)), [14, 15, 16])
+      }))
+  })
 })