From f84bb1cbf177924dfdaf45adba64864cc11f3756 Mon Sep 17 00:00:00 2001 From: Prachi Gauriar Date: Tue, 30 Jun 2026 11:50:53 -0400 Subject: [PATCH] Add CurrentValueMulticaster --- CHANGELOG.md | 7 + .../DevFoundation/Extensions/Date+Unix.swift | 4 +- .../CurrentValueMulticaster.swift | 145 +++++++++ .../Utility Types/ObservableReference.swift | 4 + .../Extensions/Date+UnixTests.swift | 4 +- .../CurrentValueMulticasterTests.swift | 288 ++++++++++++++++++ 6 files changed, 448 insertions(+), 4 deletions(-) create mode 100644 Sources/DevFoundation/Utility Types/CurrentValueMulticaster.swift create mode 100644 Tests/DevFoundationTests/Utility Types/CurrentValueMulticasterTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index ffea79a..597a03f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # DevFoundation Changelog +## 1.9.0: June 30, 2026 + +This release adds a new utility type called `CurrentValueMulticaster`. The type, similar to +`ObservableReference`, stores a value and multicasts value updates via its `values()` async +sequence. This type is preferable to use over `ObservableReference` when you’re not using SwiftUI or +other observable types. + ## 1.8.0: January 13, 2026 diff --git a/Sources/DevFoundation/Extensions/Date+Unix.swift b/Sources/DevFoundation/Extensions/Date+Unix.swift index 3c588fa..5062c19 100644 --- a/Sources/DevFoundation/Extensions/Date+Unix.swift +++ b/Sources/DevFoundation/Extensions/Date+Unix.swift @@ -34,7 +34,7 @@ extension Date { /// The date represented as whole seconds since January 1, 1970 at 00:00:00 UTC. /// - /// The fractional seconds component of the date's time interval is truncated. + /// The fractional seconds component of the date’s time interval is truncated. public var secondsSince1970: Int64 { return Int64(timeIntervalSince1970) } @@ -42,7 +42,7 @@ extension Date { /// The date represented as whole milliseconds since January 1, 1970 at 00:00:00 UTC. /// - /// The sub-millisecond component of the date's time interval is truncated. + /// The sub-millisecond component of the date’s time interval is truncated. public var millisecondsSince1970: Int64 { return Int64(timeIntervalSince1970 * 1000) } diff --git a/Sources/DevFoundation/Utility Types/CurrentValueMulticaster.swift b/Sources/DevFoundation/Utility Types/CurrentValueMulticaster.swift new file mode 100644 index 0000000..b4b2706 --- /dev/null +++ b/Sources/DevFoundation/Utility Types/CurrentValueMulticaster.swift @@ -0,0 +1,145 @@ +// +// CurrentValueMulticaster.swift +// DevFoundation +// +// Created by Prachi Gauriar on 6/30/26. +// + +import Foundation +import Synchronization + +/// A reference that holds a current value and multicasts it, plus every subsequent update, to many independent async +/// sequence consumers. +/// +/// `CurrentValueMulticaster` is the async sequence analogue of Combine’s `CurrentValueSubject`. It holds a current +/// value that callers can read and write through ``value``, and it vends async sequences via ``values()``. Each vended +/// sequence immediately emits the current value, then every subsequent update. +/// +/// Unlike ``ObservableReference``, which is best consumed using the Observation framework and SwiftUI, this type is +/// designed for publishing a current value from an `actor` or to `AsyncSequence` consumers. A subscriber’s sequence is +/// registered, and the current value emitted to it, under the same lock that updates take, so no update can slip into +/// the gap between reading the current value and subscribing. Publishing is synchronous and never suspends, so an +/// owning actor never waits on a slow consumer, and each consumer has its own buffer, so consumers never delay one +/// another. +public final class CurrentValueMulticaster: Sendable where Element: Sendable { + /// A policy that determines how a consumer’s sequence buffers values that it has not yet received. + /// + /// Each consumer buffers independently. The default, ``bufferingNewest(_:)`` with a count of `1`, yields + /// current-value semantics: if updates arrive before a consumer pulls, only the most recent is retained. + public enum BufferingPolicy: Sendable, Hashable { + /// Buffers an unbounded number of values. + case unbounded + + /// Buffers the specified number of newest values, dropping older values to make room. + case bufferingNewest(Int) + + /// Buffers the specified number of oldest values, dropping newer values once full. + case bufferingOldest(Int) + } + + + /// The mutable state that the multicaster guards behind a single lock. + private struct State { + /// The current value. + var currentValue: Element + + /// The continuations of the multicaster’s active consumer sequences, keyed by a per-consumer identifier. + var continuations: [UUID: AsyncStream.Continuation] + } + + + /// The buffering policy applied to each consumer’s sequence. + private let bufferingPolicy: BufferingPolicy + + /// A mutex that synchronizes access to the multicaster’s state. + private let state: Mutex + + + /// Creates a new multicaster with the specified initial value. + /// + /// - Parameters: + /// - initialValue: The initial value that the multicaster holds and emits to new consumers. + /// - bufferingPolicy: The policy applied to each consumer’s sequence. The default, `.bufferingNewest(1)`, + /// yields current-value semantics. + public init( + _ initialValue: Element, + bufferingPolicy: BufferingPolicy = .bufferingNewest(1), + ) { + self.bufferingPolicy = bufferingPolicy + self.state = .init(State(currentValue: initialValue, continuations: [:])) + } + + + deinit { + state.withLock { (state) in + for (_, continuation) in state.continuations { + continuation.finish() + } + } + } + + + /// The multicaster’s current value. + /// + /// Reading returns the current value. Assigning updates it and synchronously emits the new value to every active + /// consumer. Assignment emits on every write, even when the new value equals the old one. + public var value: Element { + get { + state.withLock(\.currentValue) + } + + set { + state.withLock { (state) in + state.currentValue = newValue + for continuation in state.continuations.values { + continuation.yield(newValue) + } + } + } + } + + + /// Returns a sequence that emits the current value immediately, then every subsequent update. + /// + /// Each call returns an independent sequence with its own buffer, governed by the multicaster’s buffering policy. + /// The sequence finishes when the consumer’s task is cancelled or when the multicaster is deallocated. + public func values() -> some AsyncSequence & Sendable { + let id = UUID() + + // `makeStream` is used rather than the closure-based `AsyncStream` initializer on purpose: that initializer’s + // build closure would capture `self` strongly and be retained for the stream’s lifetime, so a consumer + // holding the stream would keep the multicaster alive — preventing deallocation and the stream from ever + // finishing. Here the only retained closure is `onTermination`, which holds `self` weakly. + let (stream, continuation) = AsyncStream.makeStream( + bufferingPolicy: bufferingPolicy.asyncStreamBufferingPolicy + ) + + continuation.onTermination = { [weak self] _ in + self?.state.withLock { $0.continuations[id] = nil } + } + + // Yield the current value and register the continuation under the same lock that updates take, so no update + // can slip into the gap between reading the current value and subscribing. + state.withLock { (state) in + continuation.yield(state.currentValue) + state.continuations[id] = continuation + } + + return stream + } +} + + +extension CurrentValueMulticaster.BufferingPolicy { + /// The equivalent `AsyncStream` buffering policy. + fileprivate var asyncStreamBufferingPolicy: AsyncStream.Continuation.BufferingPolicy { + switch self { + case .unbounded: + .unbounded + case .bufferingNewest(let count): + .bufferingNewest(count) + case .bufferingOldest(let count): + .bufferingOldest(count) + } + } +} diff --git a/Sources/DevFoundation/Utility Types/ObservableReference.swift b/Sources/DevFoundation/Utility Types/ObservableReference.swift index 01990a1..6a57471 100644 --- a/Sources/DevFoundation/Utility Types/ObservableReference.swift +++ b/Sources/DevFoundation/Utility Types/ObservableReference.swift @@ -9,6 +9,10 @@ import Foundation import Synchronization /// A reference whose value changes can be observed. +/// +/// `ObservableReference` is best consumed using the Observation framework and SwiftUI. To publish a current value +/// from an `actor`, or to consume value changes as an `AsyncSequence`, prefer ``CurrentValueMulticaster``, which +/// avoids the timing pitfalls of bridging Observation into an async sequence. @Observable public final class ObservableReference: Sendable where Value: Sendable { /// A mutex that synchronizes access to the reference’s value. diff --git a/Tests/DevFoundationTests/Extensions/Date+UnixTests.swift b/Tests/DevFoundationTests/Extensions/Date+UnixTests.swift index fa0d67c..537fd6e 100644 --- a/Tests/DevFoundationTests/Extensions/Date+UnixTests.swift +++ b/Tests/DevFoundationTests/Extensions/Date+UnixTests.swift @@ -43,7 +43,7 @@ struct Date_UnixTests: RandomValueGenerating { #expect( date.timeIntervalSince1970.isApproximatelyEqual( to: Float64(milliseconds) / 1000, - absoluteTolerance: 0.0001 + absoluteTolerance: 0.0001, ) ) } @@ -63,7 +63,7 @@ struct Date_UnixTests: RandomValueGenerating { #expect( date.timeIntervalSince1970.isApproximatelyEqual( to: milliseconds / 1000, - absoluteTolerance: 0.0001 + absoluteTolerance: 0.0001, ) ) } diff --git a/Tests/DevFoundationTests/Utility Types/CurrentValueMulticasterTests.swift b/Tests/DevFoundationTests/Utility Types/CurrentValueMulticasterTests.swift new file mode 100644 index 0000000..5e85be1 --- /dev/null +++ b/Tests/DevFoundationTests/Utility Types/CurrentValueMulticasterTests.swift @@ -0,0 +1,288 @@ +// +// CurrentValueMulticasterTests.swift +// DevFoundation +// +// Created by Prachi Gauriar on 6/30/26. +// + +import DevFoundation +import DevTesting +import Foundation +import Testing + +struct CurrentValueMulticasterTests: RandomValueGenerating { + var randomNumberGenerator = makeRandomNumberGenerator() + + + // MARK: - value + + @Test + mutating func valueReturnsInitialValue() { + // set up a multicaster with a random initial value + let initialValue = randomInt(in: .min ... .max) + let multicaster = CurrentValueMulticaster(initialValue) + + // exercise / expect the getter returns the initial value + #expect(multicaster.value == initialValue) + } + + + @Test + mutating func settingValueUpdatesValue() { + // set up a multicaster with a random initial value + let multicaster = CurrentValueMulticaster(randomInt(in: .min ... .max)) + + // exercise by assigning a new value + let newValue = randomInt(in: .min ... .max) + multicaster.value = newValue + + // expect the getter reflects the new value + #expect(multicaster.value == newValue) + } + + + // MARK: - values() + + @Test + mutating func valuesEmitsCurrentValueOnSubscribe() async { + // set up a multicaster with a random initial value + let initialValue = randomInt(in: .min ... .max) + let multicaster = CurrentValueMulticaster(initialValue) + + // exercise by taking the first element of a fresh sequence + var firstValue: Int? + for await value in multicaster.values() { + firstValue = value + break + } + + // expect the first element is the current value + #expect(firstValue == initialValue) + } + + + @Test + mutating func valuesEmitsSubsequentUpdatesInOrder() async { + // set up an unbounded multicaster so no updates are dropped + let initialValue = randomInt(in: .min ... .max) + let updates = Array(count: 5) { randomInt(in: .min ... .max) } + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .unbounded) + + // exercise by subscribing, then applying every update + let values = multicaster.values() + for update in updates { + multicaster.value = update + } + + // expect the consumer observes the current value followed by every update, in order + var received: [Int] = [] + for await value in values { + received.append(value) + if received.count == updates.count + 1 { + break + } + } + + #expect(received == [initialValue] + updates) + } + + + @Test + mutating func lateSubscriberReceivesLatestValueThenFutureUpdates() async { + // set up an unbounded multicaster and apply several updates before subscribing + let initialValue = randomInt(in: .min ... .max) + let earlyUpdates = Array(count: 3) { randomInt(in: .min ... .max) } + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .unbounded) + for update in earlyUpdates { + multicaster.value = update + } + + // exercise by subscribing late, then applying one more update + let values = multicaster.values() + let laterUpdate = randomInt(in: .min ... .max) + multicaster.value = laterUpdate + + // expect the late subscriber sees the most recent value first, then the future update + var received: [Int] = [] + for await value in values { + received.append(value) + if received.count == 2 { + break + } + } + + #expect(received == [earlyUpdates.last, laterUpdate]) + } + + + @Test + mutating func updateAfterSubscribeBeforeFirstPullIsNotLost() async { + // set up the motivating race: subscribe, then update before the first pull + let initialValue = randomInt(in: .min ... .max) + let newValue = randomInt(in: .min ... .max) + + // with an unbounded policy, both the value-at-subscribe and the update are observed in order + let unboundedMulticaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .unbounded) + let unboundedValues = unboundedMulticaster.values() + unboundedMulticaster.value = newValue + + var unboundedReceived: [Int] = [] + for await value in unboundedValues { + unboundedReceived.append(value) + if unboundedReceived.count == 2 { + break + } + } + + #expect(unboundedReceived == [initialValue, newValue]) + + // with the default newest-1 policy, the update is still observed — it is never lost + let newestMulticaster = CurrentValueMulticaster(initialValue) + let newestValues = newestMulticaster.values() + newestMulticaster.value = newValue + + var newestFirstValue: Int? + for await value in newestValues { + newestFirstValue = value + break + } + + #expect(newestFirstValue == newValue) + } + + + @Test + mutating func multipleConsumersEachReceiveEveryUpdate() async { + // set up an unbounded multicaster and subscribe several consumers up front + let initialValue = randomInt(in: .min ... .max) + let updates = Array(count: 4) { randomInt(in: .min ... .max) } + let expectedValues = [initialValue] + updates + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .unbounded) + + let consumerCount = 3 + let sequences = (0 ..< consumerCount).map { _ in multicaster.values() } + + // exercise by applying every update after all consumers have subscribed + for update in updates { + multicaster.value = update + } + + // expect every consumer observes the current value followed by every update, in order + for sequence in sequences { + var received: [Int] = [] + for await value in sequence { + received.append(value) + if received.count == expectedValues.count { + break + } + } + + #expect(received == expectedValues) + } + } + + + // MARK: - Lifetime + + @Test + mutating func deallocatingMulticasterFinishesConsumerStreams() async { + // set up a multicaster referenced only by a local optional + let initialValue = randomInt(in: .min ... .max) + var multicaster: CurrentValueMulticaster? = CurrentValueMulticaster(initialValue) + + // subscribe a consumer that counts the values it receives before its stream finishes + let values = multicaster!.values() + let consumer = Task { + var receivedCount = 0 + for await _ in values { + receivedCount += 1 + } + return receivedCount + } + + // exercise by releasing the only strong reference to the multicaster + multicaster = nil + + // expect the consumer’s stream finishes after delivering the buffered current value + let receivedCount = await consumer.value + #expect(receivedCount == 1) + } + + + // MARK: - Buffering policy + + @Test + mutating func bufferingNewestKeepsLatestValue() async { + // set up a newest-1 multicaster and subscribe before applying updates + let initialValue = randomInt(in: .min ... .max) + let updates = Array(count: 3) { randomInt(in: .min ... .max) } + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .bufferingNewest(1)) + + // exercise by applying every update without consuming, so only the newest is retained + let values = multicaster.values() + for update in updates { + multicaster.value = update + } + + // expect the first value the slow consumer pulls is the most recent + var firstValue: Int? + for await value in values { + firstValue = value + break + } + + #expect(firstValue == updates.last) + } + + + @Test + mutating func unboundedBuffersAllValues() async { + // set up an unbounded multicaster and subscribe before applying updates + let initialValue = randomInt(in: .min ... .max) + let updates = Array(count: 3) { randomInt(in: .min ... .max) } + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .unbounded) + + // exercise by applying every update without consuming + let values = multicaster.values() + for update in updates { + multicaster.value = update + } + + // expect the slow consumer pulls every value in order + var received: [Int] = [] + for await value in values { + received.append(value) + if received.count == updates.count + 1 { + break + } + } + + #expect(received == [initialValue] + updates) + } + + + @Test + mutating func bufferingOldestKeepsEarliestValues() async { + // set up an oldest-2 multicaster and subscribe before applying updates + let initialValue = randomInt(in: .min ... .max) + let updates = Array(count: 3) { randomInt(in: .min ... .max) } + let multicaster = CurrentValueMulticaster(initialValue, bufferingPolicy: .bufferingOldest(2)) + + // exercise by applying every update without consuming, so only the oldest two are retained + let values = multicaster.values() + for update in updates { + multicaster.value = update + } + + // expect the slow consumer pulls the current value and the first update, in order + var received: [Int] = [] + for await value in values { + received.append(value) + if received.count == 2 { + break + } + } + + #expect(received == [initialValue, updates.first]) + } +}