Replace AsyncPassthroughSubject from AsyncExtension

Author: intitni

Core/Sources/Service/RealtimeSuggestionController.swift

@@ -65,8 +65,9 @@ public actor RealtimeSuggestionController {
                 )
             }
 
-            let valueChange = notificationsFromEditor.filter { $0.kind == .valueChanged }
-            let selectedTextChanged = notificationsFromEditor
+            let valueChange = await notificationsFromEditor.notifications()
+                .filter { $0.kind == .valueChanged }
+            let selectedTextChanged = await notificationsFromEditor.notifications()
                 .filter { $0.kind == .selectedTextChanged }
 
             await withTaskGroup(of: Void.self) { [weak self] group in

Core/Sources/SuggestionWidget/WidgetWindowsController.swift

@@ -288,7 +288,7 @@ private extension WidgetWindowsController {
             await windows.orderFront()
 
             let documentURL = await MainActor.run { store.withState { $0.focusingDocumentURL } }
-            for await notification in notifications {
+            for await notification in await notifications.notifications() {
                 try Task.checkCancellation()
 
                 /// Hide the widgets before switching to another window/editor
@@ -337,9 +337,9 @@ private extension WidgetWindowsController {
     func observe(to editor: SourceEditor) {
         observeToFocusedEditorTask?.cancel()
         observeToFocusedEditorTask = Task {
-            let selectionRangeChange = editor.axNotifications
+            let selectionRangeChange = await editor.axNotifications.notifications()
                 .filter { $0.kind == .selectedTextChanged }
-            let scroll = editor.axNotifications
+            let scroll = await editor.axNotifications.notifications()
                 .filter { $0.kind == .scrollPositionChanged }
 
             if #available(macOS 13.0, *) {

Pro

@@ -43,6 +43,7 @@ let package = Package(
         ),
         .library(name: "GitIgnoreCheck", targets: ["GitIgnoreCheck"]),
         .library(name: "DebounceFunction", targets: ["DebounceFunction"]),
+        .library(name: "AsyncPassthroughSubject", targets: ["AsyncPassthroughSubject"]),
     ],
     dependencies: [
         // A fork of https://github.com/aespinilla/Tiktoken to allow loading from local files.
@@ -64,7 +65,6 @@ let package = Package(
         .package(url: "https://github.com/GottaGetSwifty/CodableWrappers", from: "2.0.7"),
         .package(url: "https://github.com/krzyzanowskim/STTextView", from: "0.8.21"),
         .package(url: "https://github.com/google/generative-ai-swift", from: "0.4.4"),
-        .package(url: "https://github.com/sideeffect-io/AsyncExtensions", from: "0.5.2"),
 
         // TreeSitter
         .package(url: "https://github.com/intitni/SwiftTreeSitter.git", branch: "main"),
@@ -173,7 +173,7 @@ let package = Package(
                 "Logger",
                 "Toast",
                 "Preferences",
-                .product(name: "AsyncExtensions", package: "AsyncExtensions"),
+                "AsyncPassthroughSubject",
                 .product(name: "AsyncAlgorithms", package: "swift-async-algorithms"),
             ]
         ),
@@ -182,6 +182,8 @@ let package = Package(
 
         .target(name: "UserDefaultsObserver"),
 
+        .target(name: "AsyncPassthroughSubject"),
+
         .target(
             name: "SharedUIComponents",
             dependencies: [

Tool/Package.swift

@@ -0,0 +1,69 @@
+import AppKit
+import Foundation
+
+/// It uses notification center to mimic the behavior of a passthrough subject.
+public actor AsyncPassthroughSubject<Element> {
+    let name: Notification.Name
+    var tasks: [AsyncStream<Element>.Continuation] = []
+
+    deinit {
+        tasks.forEach { $0.finish() }
+    }
+    
+    public init() {
+        name = NSNotification.Name(
+            "AsyncPassthroughSubject-\(UUID().uuidString)-\(String(describing: Element.self))"
+        )
+    }
+
+    public func notifications() -> AsyncStream<Element> {
+        AsyncStream { [weak self, name] continuation in
+            let task = Task { [weak self] in
+                await self?.storeContinuation(continuation)
+                let notifications = NSWorkspace.shared.notificationCenter.notifications(named: name)
+                    .compactMap {
+                        $0.object as? Element
+                    }
+                for await notification in notifications {
+                    try Task.checkCancellation()
+                    guard self != nil else {
+                        continuation.finish()
+                        return
+                    }
+                    continuation.yield(notification)
+                }
+            }
+            
+            continuation.onTermination = { termination in
+                task.cancel()
+            }
+        }
+    }
+
+    nonisolated
+    public func send(_ element: Element) {
+        Task { await _send(element) }
+    }
+    
+    func _send(_ element: Element) {
+        NSWorkspace.shared.notificationCenter.post(name: name, object: element)
+    }
+    
+    func storeContinuation(_ continuation: AsyncStream<Element>.Continuation) {
+        tasks.append(continuation)
+    }
+    
+    nonisolated
+    public func finish() {
+        Task { await _finish() }
+    }
+    
+    func _finish() {
+        let tasks = self.tasks
+        self.tasks = []
+        for task in tasks {
+            task.finish()
+        }
+    }
+}
+

Tool/Sources/AsyncPassthroughSubject/AsyncPassthroughSubject.swift

@@ -1,5 +1,5 @@
 import AppKit
-import AsyncExtensions
+import AsyncPassthroughSubject
 import AXExtension
 import AXNotificationStream
 import Combine
@@ -123,7 +123,7 @@ public final class XcodeAppInstanceInspector: AppInstanceInspector {
     private var focusedWindowObservations = Set<AnyCancellable>()
 
     deinit {
-        axNotifications.send(.finished)
+        axNotifications.finish()
         for task in longRunningTasks { task.cancel() }
     }
 

Tool/Sources/XcodeInspector/Apps/XcodeAppInstanceInspector.swift

@@ -1,5 +1,5 @@
 import AppKit
-import AsyncExtensions
+import AsyncPassthroughSubject
 import AXNotificationStream
 import Foundation
 import Logger

Tool/Sources/XcodeInspector/SourceEditor.swift

@@ -286,7 +286,7 @@ public final class XcodeInspector: ObservableObject {
 
         setFocusedElement()
         let focusedElementChanged = Task { @XcodeInspectorActor in
-            for await notification in xcode.axNotifications {
+            for await notification in await xcode.axNotifications.notifications() {
                 if notification.kind == .focusedUIElementChanged {
                     try Task.checkCancellation()
                     setFocusedElement()
@@ -301,7 +301,7 @@ public final class XcodeInspector: ObservableObject {
         {
             let malfunctionCheck = Task { @XcodeInspectorActor [weak self] in
                 if #available(macOS 13.0, *) {
-                    let notifications = xcode.axNotifications.filter {
+                    let notifications = await xcode.axNotifications.notifications().filter {
                         $0.kind == .uiElementDestroyed
                     }.debounce(for: .milliseconds(1000))
                     for await _ in notifications {

Tool/Sources/XcodeInspector/XcodeInspector.swift

@@ -1,5 +1,5 @@
 import AppKit
-import AsyncExtensions
+import AsyncPassthroughSubject
 import AXExtension
 import Combine
 import Foundation
@@ -19,11 +19,6 @@ public final class WorkspaceXcodeWindowInspector: XcodeWindowInspector {
     @Published var workspaceURL: URL = .init(fileURLWithPath: "/")
     @Published var projectRootURL: URL = .init(fileURLWithPath: "/")
     private var focusedElementChangedTask: Task<Void, Error>?
-    let axNotifications: AsyncPassthroughSubject<XcodeAppInstanceInspector.AXNotification>
-
-    deinit {
-        focusedElementChangedTask?.cancel()
-    }
 
     public func refresh() {
         Task { @XcodeInspectorActor in updateURLs() }
@@ -35,7 +30,6 @@ public final class WorkspaceXcodeWindowInspector: XcodeWindowInspector {
         axNotifications: AsyncPassthroughSubject<XcodeAppInstanceInspector.AXNotification>
     ) {
         self.app = app
-        self.axNotifications = axNotifications
         super.init(uiElement: uiElement)
 
         focusedElementChangedTask = Task { [weak self, axNotifications] in
@@ -51,7 +45,7 @@ public final class WorkspaceXcodeWindowInspector: XcodeWindowInspector {
                 }
 
                 group.addTask { [weak self] in
-                    for await notification in axNotifications {
+                    for await notification in await axNotifications.notifications() {
                         guard notification.kind == .focusedUIElementChanged else { continue }
                         guard let self else { return }
                         try Task.checkCancellation()