Question about EventSequence

[Wouter01]

Hi,

I was refactoring some of my code and stumbled upon a weird issue: I couldn't get events from eventSequence (in RestartingServer) anymore. When I added a delay, it works fine. Turns out, it was because I accidentally started the task containing the for await _ in eventSequence twice, and cancelled the first one. Cancelling a task will cancel the streams inside it too, therefore cancelling eventSequence. A cancelled stream cannot be restarted, removing the ability to get any more events from the language server, as I cannot find a way to make a new EventSequence with the public API.

I was wondering if this falls under "intended behavior"? Could it be beneficial to have the ability to create multiple EventSequence instances, so a sequence can be cancelled and multiple tasks can get the events?

[mattmassicotte]

This wasn't the intentional behavior. What do you think should happen here? And, do you know why adding a delay helped?

If you are talking about multiple consumers of eventSequence, that is, sadly, not supported by AsyncSequence today. But, it is desperately needed, in my opinion. https://github.com/apple/swift-async-algorithms/issues/110

[Wouter01]

In my case, adding the delay mostly worked because then I would only start listening to the stream once, so I wouldn't have the issue.

You're right that it isn't possible at the moment to broadcast the events. I guess it could be done by using another method, however this isn't a huge issue so I don't think it's needed

[mattmassicotte]

The behavior isn't good, but it's more about how AsyncSequence works than this library. If you do not agree, please re-open!

[malhal]

The problem is eventSequence is an instance var of a class but it should be a computed var so it can be created by multiple clients, since that is the how they are designed to work. E.g. you can create a streams for NotificationCenter notifications all over your app and filter or process each one how you like, you aren't limited to one notification stream. I noticed the CoreLocation team has made the same design flaw in their CLMonitor events stream.

[mattmassicotte]

Ok, I'm listening! If eventSequence creates a new sequence on invocation, it does fix this issue client side. Right now the design uses AsyncStream, so there would have to a collection of continuations somewhere. How should that work?

[malhal]

Continuations are for bridging between delegates/closures to async. Since you say your design already is async perhaps you could have a collection of Channels, one per client:

Then you might also like to wrap each Channel in an AsyncSequence similar to how streams are usually wrapped in sequences for public facing APIs: https://forums.swift.org/t/how-do-you-completely-cancel-an-asyncstream/53733/2

I'll give this design a try tomorrow to see if it fixes the issue with CLMonitor and report back.

[malhal]

Here is an implementation that did choose to use a collection of continuations: https://github.com/reddavis/Asynchrone/blob/main/Sources/Asynchrone/Sequences/SharedAsyncSequence.swift

[malhal]

I've looked into this and learned the best way to implement multiple consumers of the eventSequence is for the client to hold an array of AsyncChannels and send the event to all of them. This design worked fine for CLMonitor:

One place:

@MainActor
class MonitorController {
    let monitor: CLMonitor
    var channels: [AsyncChannel<CLMonitor.Event>] = []

    init() async {
        monitor = await CLMonitor("SampleMonitor")
    }
    
    func start() async {
        print("Started")
        do {
            for try await event in await monitor.events {
                for channel in channels {
                    await channel.send(event)
                }
            }
        }catch {}
        print("Ended")
    }

Many places:

        .task {
            print("started")
            let channel = AsyncChannel<CLMonitor.Event>()
            monitorController.channels.append(channel)
            for await event in channel {
                print("event")
            }
            monitorController.channels.removeAll { e in
                e === channel
            }
        }

Since AsyncChannel is a class, it's best for the client to manage that reference. Trying from inside the server is error-prone and might be an anti-pattern anyway. So you can probably just close this issue again 😉

[mattmassicotte]

Thanks for all your investigations and input here. I do appreciate it!