Making AsyncSequence Sendable (Debounce on AsyncSequence)

Using Swift
1

Hi,

Overview

  • I would like to use debounce on AsyncSequence
  • I am using AsyncAlgorithms
  • I am using Swift 6.4

Questions

  • Can I make cars (AsyncSequence) Sendable?
  • Why is AsyncSequence<Car, Never> not Sendable when Car is Sendable?
  • How can I make this work?

Code

import AsyncAlgorithms

struct Car: Sendable {
    init(notification: Notification) {}
}

let testNotificationName = Notification.Name(rawValue: "Test")

let notifications = NotificationCenter.default.notifications(named: testNotificationName)

// Conformance of 'Notification' to 'Sendable' is unavailable in iOS
let cars: some AsyncSequence<Car, Never> & Sendable = notifications.map { Car(notification: $0) }

// How can I make cars Sendable?
for await car in cars.debounce(for: .seconds(2)) {
    
}
1 Like
2

Notifications isn't Sendable because Notification isn't Sendable.

AsyncMapSequence<Notifications, Car> isn't Sendable because it contains a Notifications, which isn't Sendable.

Technically, there's "nothing" you can do about this.

If you can use the latest & greatest stuffs, this year's OSes seem to have support for SF-0011 which has some new stuff for dealing with notifications in concurrency-safe ways.

If you don't care about the Notification itself, you can reimplement Notifications easily enough to be Sendable and emit any Sendable data you can extract from the Notification.

2 Likes
3

Thanks @KeithBauerANZ for clarifying on that.

For now will use Combine

Is the quoted message applicable now or do you mean it is possible after SF-0011?

4

It is possible now.

Note that anything that sends a notification using Combine will be just as wrong, the compiler just won't yell at you about it ;p

1 Like
5

@KeithBauerANZ would it be possible to give some guidance on how Notifications can be implemented?

Presently my solution using Combine is as follows, do you think the below is problematic?

import Foundation
import Combine

struct Car: Sendable {
    init(notification: Notification) {}
}

let testNotificationName = Notification.Name(rawValue: "Test")

let notifications = NotificationCenter.default.notifications(named: testNotificationName)

let cars = NotificationCenter.default.publisher(for: testNotificationName)
    .receive(on: DispatchQueue.main)
    .compactMap { Car(notification: $0) }
    .debounce(for: .seconds(2), scheduler: DispatchQueue.main)
    .values

for await car in cars {
    
}
6

Do not use Notification, but AsyncMessage.

1 Like
7

Thanks a lot @Akazm and @KeithBauerANZ

Wow this is super cool!

My real world scenario is I was observing some notifications posted by Apple framework. Those specific notifications have Async message from OS 27 onwards.

Question

  • For framework notifications which are unavailable on current OS version, is it ok to fall back to my Combine implementation?
8

Bridging Combine and AsyncSequence is not a trivial task. Claude & Co. might help you, but be prepared for wasting tokens on this.

Above that, AsyncMessage is available from iOS26. You can always implement your own AsyncMessage conformance in order to retroactively subscribe to notifications that don't come with an AsyncMessage implementation by default.

For this, implement makeMessage and ensure the notification name matches.

1 Like
9

thanks a ton to @Akazm and @KeithBauerANZ

Works really well, was very helpful!

1 Like
10

Manually implementing the Notifications stream would just be wrapping the closure-based NotificationCenter API with something like AsyncStream and an immediate map closure that transforms the Notification before it enters the stream.

1 Like
11

Thanks @Jon_Shier for the suggestion

12

It's actually pretty well explained in Foundation documentation too: notifications(named:object:) | Apple Developer Documentation.

13

Which is the approach taken here, but since debounce requires both the parent sequence and the value to be Sendable, mapping out to a custom type doesn't actually help. The mapping needs to be done before the AsyncSequence is created in the first place, so all of them have the opportunity to be Sendable before they reach debounce.

14

If testNotificationName notifications are always sent on the main queue, that's fine (but your .receive(on:) is redundant. If they're not, then you're sending their Notification from wherever they're posted to the main queue, which is a bug.

untested:

struct NotificationEvents<Element: Sendable>: AsyncSequence, Sendable {

    @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
    typealias Failure = Never

    final class Subscriber: NSObject {
        let continuation: AsyncStream<Element>.Continuation
        let extract: @Sendable (Notification) -> Element

        init(
            continuation: AsyncStream<Element>.Continuation,
            extract: @Sendable @escaping (Notification) -> Element
        ) {
            self.continuation = continuation
            self.extract = extract
        }

        @objc
        func notify(_ notification: Notification) {
            continuation.yield(extract(notification))
        }
    }

    struct AsyncIterator: AsyncIteratorProtocol {
        var subscriber: Subscriber
        var inner: AsyncStream<Element>.AsyncIterator

        @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
        mutating func next(isolation actor: isolated (any Actor)?) async -> Element? {
            await inner.next(isolation: actor)
        }

        mutating func next() async -> Element? {
            await inner.next()
        }
    }

    var name: Notification.Name
    var extract: @Sendable (Notification) -> Element

    init(
        name: Notification.Name,
        extract: @escaping @Sendable (Notification) -> Element
    ) {
        self.name = name
        self.extract = extract
    }

    func makeAsyncIterator() -> AsyncIterator {
        let (stream, continuation) = AsyncStream.makeStream(of: Element.self, bufferingPolicy: .bufferingNewest(8))
        let subscriber = Subscriber(continuation: continuation, extract: extract)
        NotificationCenter.default.addObserver(
            subscriber,
            selector: #selector(Subscriber.notify(_:)),
            name: name,
            object: nil
        )
        return AsyncIterator(
            subscriber: subscriber,
            inner: stream.makeAsyncIterator()
        )
    }
}
2 Likes
15

Thanks a lot @KeithBauerANZ, I will try out.