How to implement SharedStateClient?

[takehilo]

@mbrandonw
I would like to know how to implement SharedStateClient you suggest in the following Discussion.
#1898 (reply in thread)

I tried implementing SharedStateClient as follows.
Since AsyncStream cannot have multiple consumers, I achieved an AsyncStream that can handle multiple consumers by using CurrentValueSubject(ref: https://forums.swift.org/t/asyncsequence-stream-version-of-passthroughsubject-or-currentvaluesubject/60395).
However, there is a problem with this implementation.
CurrentValueSubject is not Sendable, so it will cause an error in future Swift updates.
I would appreciate it if you could share any ideas for better implementations.

struct SharedStateClient {
    var value: @Sendable () -> SharedState
    var setValue: @Sendable (SharedState) -> Void
    var values: @Sendable () -> AsyncStream<SharedState>

    init(
        value: @Sendable @escaping () -> SharedState,
        setValue: @Sendable @escaping (SharedState) -> Void,
        values: @Sendable @escaping () -> AsyncStream<SharedState>
    ) {
        self.value = value
        self.setValue = setValue
        self.values = values
    }
}

extension SharedStateClient {
    static func live() -> Self {
        let subject = CurrentValueSubject<SharedState, Never>(.init(text: "first"))

        return .init(
            value: {
                subject.value
            },
            setValue: {
                subject.send($0)
            },
            values: {
                AsyncStream { continuation in
                    let cancellable = subject
                        .dropFirst()
                        .removeDuplicates()
                        .sink { value in
                            continuation.yield(value)
                    }

                    continuation.onTermination = { _ in
                        cancellable.cancel()
                    }
                }
            }
        )
    }
}

[fabstu]

Dunno how others do it but this is my handwavy approach:

For me SharedStateClient is a class and the value is just a variable inside the class. SharedStateClient stores a watchers-array with a watcher containing an ID and the continuation of the AsyncStream it was created for.

On creation: directly before .onTermination the new watcher is created and inserted into the watchers-array. On deletion: onTermination is called automatically so delete the watcher (we have the ID locally since it was created here) from the watchers-array. On update, each watchers' continuation.yield is called with the new value.

I feel like I'm missing some construct though to simplify.

[coenttb]

Also interested in tips for best-practice for SharedStateClient.

[takehilo]

I made SharedStateClient Sendable using LockIsolated and UncheckedSendable.

struct SharedStateClient: Sendable {
    var value: @Sendable () -> SharedState
    var setValue: @Sendable (SharedState) -> Void
    var values: @Sendable () -> AsyncStream<SharedState>

    init(
        value: @Sendable @escaping () -> SharedState,
        setValue: @Sendable @escaping (SharedState) -> Void,
        values: @Sendable @escaping () -> AsyncStream<SharedState>
    ) {
        self.value = value
        self.setValue = setValue
        self.values = values
    }
}

extension SharedStateClient {
    static func live() -> Self {
        let subject = LockIsolated<CurrentValueSubject<SharedState, Never>>(.init(.init(text: "first")))

        return .init(
            value: {
                subject.value
            },
            setValue: { value in
                subject.withValue { $0.send(value) }
            },
            values: {
                AsyncStream { continuation in
                    subject.withValue {
                        let cancellable = $0
                            .removeDuplicates()
                            .sink { value in
                                continuation.yield(value)
                            }

                        continuation.onTermination = { [cancellable = UncheckedSendable(cancellable)] _ in
                            cancellable.value.cancel()
                        }
                    }
                }
            }
        )
    }
}