Replace custom Core Data observer with Combine publisher (#64)

* Replace custom Core Data observer with Combine publisher, thanks to @10nas 

* Replace custom notification with combine publisher

* Replace custom notification with combine publisher for FPU

* Cleanup

FreeAPS/Sources/APS/Storage/CarbsStorage.swift

@@ -1,3 +1,4 @@
+import Combine
 import CoreData
 import Foundation
 import SwiftDate
@@ -8,7 +9,9 @@ protocol CarbsObserver {
 }
 
 protocol CarbsStorage {
+    var updatePublisher: AnyPublisher<Void, Never> { get }
     func storeCarbs(_ carbs: [CarbsEntry], areFetchedFromRemote: Bool) async
+    func deleteCarbs(_ treatmentObjectID: NSManagedObjectID) async
     func syncDate() -> Date
     func recent() -> [CarbsEntry]
     func getCarbsNotYetUploadedToNightscout() async -> [NightscoutTreatment]
@@ -24,6 +27,12 @@ final class BaseCarbsStorage: CarbsStorage, Injectable {
 
     let coredataContext = CoreDataStack.shared.newTaskContext()
 
+    private let updateSubject = PassthroughSubject<Void, Never>()
+
+    var updatePublisher: AnyPublisher<Void, Never> {
+        updateSubject.eraseToAnyPublisher()
+    }
+
     init(resolver: Resolver) {
         injectServices(resolver)
     }
@@ -226,8 +235,8 @@ final class BaseCarbsStorage: CarbsStorage, Injectable {
                 try self.coredataContext.execute(batchInsert)
                 debugPrint("Carbs Storage: \(DebuggingIdentifiers.succeeded) saved fpus to core data")
 
-                // Send notification for triggering a fetch in Home State Model to update the FPU Array
-                Foundation.NotificationCenter.default.post(name: .didPerformBatchInsert, object: nil)
+                // Notify subscriber in Home State Model to update the FPU Array
+                self.updateSubject.send(())
             } catch {
                 debugPrint("Carbs Storage: \(DebuggingIdentifiers.failed) error while saving fpus to core data")
             }
@@ -242,6 +251,53 @@ final class BaseCarbsStorage: CarbsStorage, Injectable {
         storage.retrieve(OpenAPS.Monitor.carbHistory, as: [CarbsEntry].self)?.reversed() ?? []
     }
 
+    func deleteCarbs(_ treatmentObjectID: NSManagedObjectID) async {
+        let taskContext = CoreDataStack.shared.newTaskContext()
+        taskContext.name = "deleteContext"
+        taskContext.transactionAuthor = "deleteCarbs"
+
+        var carbEntry: CarbEntryStored?
+
+        await taskContext.perform {
+            do {
+                carbEntry = try taskContext.existingObject(with: treatmentObjectID) as? CarbEntryStored
+                guard let carbEntry = carbEntry else {
+                    debugPrint("Carb entry for batch delete not found. \(DebuggingIdentifiers.failed)")
+                    return
+                }
+
+                if carbEntry.isFPU, let fpuID = carbEntry.fpuID {
+                    // fetch request for all carb entries with the same id
+                    let fetchRequest: NSFetchRequest<NSFetchRequestResult> = CarbEntryStored.fetchRequest()
+                    fetchRequest.predicate = NSPredicate(format: "fpuID == %@", fpuID as CVarArg)
+
+                    // NSBatchDeleteRequest
+                    let deleteRequest = NSBatchDeleteRequest(fetchRequest: fetchRequest)
+                    deleteRequest.resultType = .resultTypeCount
+
+                    // execute the batch delete request
+                    let result = try taskContext.execute(deleteRequest) as? NSBatchDeleteResult
+                    debugPrint("\(DebuggingIdentifiers.succeeded) Deleted \(result?.result ?? 0) items with FpuID \(fpuID)")
+
+                    // Notifiy subscribers of the batch delete
+                    self.updateSubject.send(())
+                } else {
+                    taskContext.delete(carbEntry)
+
+                    guard taskContext.hasChanges else { return }
+                    try taskContext.save()
+
+                    debugPrint(
+                        "Data Table State: \(#function) \(DebuggingIdentifiers.succeeded) deleted carb entry from core data"
+                    )
+                }
+
+            } catch {
+                debugPrint("\(DebuggingIdentifiers.failed) Error deleting carb entry: \(error.localizedDescription)")
+            }
+        }
+    }
+
     func deleteCarbs(at uniqueID: String, fpuID: String, complex: Bool) {
         processQueue.sync {
             var allValues = storage.retrieve(OpenAPS.Monitor.carbHistory, as: [CarbsEntry].self) ?? []

FreeAPS/Sources/APS/Storage/DeterminationStorage.swift

@@ -1,3 +1,4 @@
+import Combine
 import CoreData
 import Foundation
 import Swinject

FreeAPS/Sources/APS/Storage/GlucoseStorage.swift

@@ -1,4 +1,5 @@
 import AVFAudio
+import Combine
 import CoreData
 import Foundation
 import SwiftDate
@@ -6,6 +7,7 @@ import SwiftUI
 import Swinject
 
 protocol GlucoseStorage {
+    var updatePublisher: AnyPublisher<Void, Never> { get }
     func storeGlucose(_ glucose: [BloodGlucose])
     func isGlucoseDataFresh(_ glucoseDate: Date?) -> Bool
     func syncDate() -> Date
@@ -26,6 +28,12 @@ final class BaseGlucoseStorage: GlucoseStorage, Injectable {
 
     let coredataContext = CoreDataStack.shared.newTaskContext()
 
+    private let updateSubject = PassthroughSubject<Void, Never>()
+
+    var updatePublisher: AnyPublisher<Void, Never> {
+        updateSubject.eraseToAnyPublisher()
+    }
+
     private enum Config {
         static let filterTime: TimeInterval = 3.5 * 60
     }
@@ -87,12 +95,10 @@ final class BaseGlucoseStorage: GlucoseStorage, Injectable {
                 // process batch insert
                 do {
                     try self.coredataContext.execute(batchInsert)
-//                    debugPrint("Glucose Storage: \(#function) \(DebuggingIdentifiers.succeeded) saved glucose to Core Data")
 
-                    // Send notification for triggering a fetch in Home State Model to update the Glucose Array
-                    /// This is necessary because changes only get merged automatically into the viewContext because of the Persistent History Tracking
-                    /// But I do not want to fetch on the Main Thread using the @FetchRequest property, I also can not use the FetchedResultsController because of the architecture of the State Model (it must inherit from BaseStateModel and therefore can not inherit from NSObject as well) and because of the fact that I am using a batch insert here there are no notifications sent from the managedObjectContext because changes are directly stored in the persistent container
-                    Foundation.NotificationCenter.default.post(name: .didPerformBatchInsert, object: nil)
+                    // Notify subscribers that there is a new glucose value
+                    // We need to do this because the due to the batch insert there is no ManagedObjectContext notification
+                    self.updateSubject.send(())
                 } catch {
                     debugPrint(
                         "Glucose Storage: \(#function) \(DebuggingIdentifiers.failed) failed to execute batch insert: \(error)"

FreeAPS/Sources/APS/Storage/PumpHistoryStorage.swift

@@ -1,3 +1,4 @@
+import Combine
 import CoreData
 import Foundation
 import LoopKit
@@ -9,6 +10,7 @@ protocol PumpHistoryObserver {
 }
 
 protocol PumpHistoryStorage {
+    var updatePublisher: AnyPublisher<Void, Never> { get }
     func storePumpEvents(_ events: [NewPumpEvent])
     func storeExternalInsulinEvent(amount: Decimal, timestamp: Date) async
     func recent() -> [PumpHistoryEvent]
@@ -22,6 +24,12 @@ final class BasePumpHistoryStorage: PumpHistoryStorage, Injectable {
     @Injected() private var broadcaster: Broadcaster!
     @Injected() private var settings: SettingsManager!
 
+    private let updateSubject = PassthroughSubject<Void, Never>()
+
+    var updatePublisher: AnyPublisher<Void, Never> {
+        updateSubject.eraseToAnyPublisher()
+    }
+
     init(resolver: Resolver) {
         injectServices(resolver)
     }
@@ -190,6 +198,8 @@ final class BasePumpHistoryStorage: PumpHistoryStorage, Injectable {
                 do {
                     guard self.context.hasChanges else { return }
                     try self.context.save()
+
+                    self.updateSubject.send(())
                     debugPrint("\(DebuggingIdentifiers.succeeded) stored pump events in Core Data")
                 } catch let error as NSError {
                     debugPrint("\(DebuggingIdentifiers.failed) failed to store pump events with error: \(error.userInfo)")
@@ -218,6 +228,8 @@ final class BasePumpHistoryStorage: PumpHistoryStorage, Injectable {
             do {
                 guard self.context.hasChanges else { return }
                 try self.context.save()
+
+                self.updateSubject.send(())
             } catch {
                 print(error.localizedDescription)
             }

FreeAPS/Sources/Modules/Bolus/BolusStateModel.swift

@@ -1,3 +1,4 @@
+import Combine
 import CoreData
 import Foundation
 import LoopKit
@@ -117,18 +118,28 @@ extension Bolus {
         let viewContext = CoreDataStack.shared.persistentContainer.viewContext
         let backgroundContext = CoreDataStack.shared.newTaskContext()
 
-        private var coreDataObserver: CoreDataObserver?
+        private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+        private var subscriptions = Set<AnyCancellable>()
 
         typealias PumpEvent = PumpEventStored.EventType
 
         override func subscribe() {
+            coreDataPublisher =
+                changedObjectsOnManagedObjectContextDidSavePublisher()
+                    .receive(on: DispatchQueue.global(qos: .background))
+                    .share()
+                    .eraseToAnyPublisher()
+            setupBolusStateConcurrently()
+        }
+
+        private func setupBolusStateConcurrently() {
             Task {
                 await withTaskGroup(of: Void.self) { group in
                     group.addTask {
-                        self.setupGlucoseNotification()
+                        self.registerHandlers()
                     }
                     group.addTask {
-                        self.registerHandlers()
+                        self.registerSubscribers()
                     }
                     group.addTask {
                         self.setupGlucoseArray()
@@ -561,33 +572,29 @@ extension Bolus.StateModel: DeterminationObserver, BolusFailureObserver {
 
 extension Bolus.StateModel {
     private func registerHandlers() {
-        coreDataObserver?.registerHandler(for: "OrefDetermination") { [weak self] in
+        coreDataPublisher?.filterByEntityName("OrefDetermination").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.setupDeterminationsArray()
                 await self.updateForecasts()
             }
-        }
+        }.store(in: &subscriptions)
 
         // Due to the Batch insert this only is used for observing Deletion of Glucose entries
-        coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+        coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
             guard let self = self else { return }
             self.setupGlucoseArray()
-        }
+        }.store(in: &subscriptions)
     }
 
-    private func setupGlucoseNotification() {
-        /// custom notification that is sent when a batch insert of glucose objects is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchInsert),
-            name: .didPerformBatchInsert,
-            object: nil
-        )
-    }
-
-    @objc private func handleBatchInsert() {
-        setupGlucoseArray()
+    private func registerSubscribers() {
+        glucoseStorage.updatePublisher
+            .receive(on: DispatchQueue.global(qos: .background))
+            .sink { [weak self] _ in
+                guard let self = self else { return }
+                self.setupGlucoseArray()
+            }
+            .store(in: &subscriptions)
     }
 }
 

FreeAPS/Sources/Modules/DataTable/DataTableStateModel.swift

@@ -10,6 +10,7 @@ extension DataTable {
         @Injected() var pumpHistoryStorage: PumpHistoryStorage!
         @Injected() var glucoseStorage: GlucoseStorage!
         @Injected() var healthKitManager: HealthKitManager!
+        @Injected() var carbsStorage: CarbsStorage!
 
         let coredataContext = CoreDataStack.shared.newTaskContext()
 
@@ -97,6 +98,7 @@ extension DataTable {
 
             var carbEntry: CarbEntryStored?
 
+            // Delete carbs or FPUs from Nightscout
             await taskContext.perform {
                 do {
                     carbEntry = try taskContext.existingObject(with: treatmentObjectID) as? CarbEntryStored
@@ -108,42 +110,23 @@ extension DataTable {
                     if carbEntry.isFPU, let fpuID = carbEntry.fpuID {
                         // Delete FPUs from Nightscout
                         self.provider.deleteCarbsFromNightscout(withID: fpuID.uuidString)
-
-                        // fetch request for all carb entries with the same id
-                        let fetchRequest: NSFetchRequest<NSFetchRequestResult> = CarbEntryStored.fetchRequest()
-                        fetchRequest.predicate = NSPredicate(format: "fpuID == %@", fpuID as CVarArg)
-
-                        // NSBatchDeleteRequest
-                        let deleteRequest = NSBatchDeleteRequest(fetchRequest: fetchRequest)
-                        deleteRequest.resultType = .resultTypeCount
-
-                        // execute the batch delete request
-                        let result = try taskContext.execute(deleteRequest) as? NSBatchDeleteResult
-                        debugPrint("\(DebuggingIdentifiers.succeeded) Deleted \(result?.result ?? 0) items with FpuID \(fpuID)")
-
-                        Foundation.NotificationCenter.default.post(name: .didPerformBatchDelete, object: nil)
                     } else {
                         // Delete carbs from Nightscout
                         if let id = carbEntry.id?.uuidString {
                             self.provider.deleteCarbsFromNightscout(withID: id)
                         }
-
-                        // Now delete carbs also from the Database
-                        taskContext.delete(carbEntry)
-
-                        guard taskContext.hasChanges else { return }
-                        try taskContext.save()
-
-                        debugPrint(
-                            "Data Table State: \(#function) \(DebuggingIdentifiers.succeeded) deleted carb entry from core data"
-                        )
                     }
 
                 } catch {
-                    debugPrint("\(DebuggingIdentifiers.failed) Error deleting carb entry: \(error.localizedDescription)")
+                    debugPrint(
+                        "\(DebuggingIdentifiers.failed) Error deleting carb entry from Nightscout: \(error.localizedDescription)"
+                    )
                 }
             }
 
+            // Delete carbs from Core Data
+            await carbsStorage.deleteCarbs(treatmentObjectID)
+
             // Perform a determine basal sync to update cob
             await apsManager.determineBasalSync()
         }

FreeAPS/Sources/Modules/Home/HomeStateModel.swift

@@ -13,6 +13,7 @@ extension Home {
         @Injected() var nightscoutManager: NightscoutManager!
         @Injected() var determinationStorage: DeterminationStorage!
         @Injected() var glucoseStorage: GlucoseStorage!
+        @Injected() var carbsStorage: CarbsStorage!
         private let timer = DispatchTimer(timeInterval: 5)
         private(set) var filteredHours = 24
         @Published var manualGlucose: [BloodGlucose] = []
@@ -90,12 +91,17 @@ extension Home {
         let context = CoreDataStack.shared.newTaskContext()
         let viewContext = CoreDataStack.shared.persistentContainer.viewContext
 
-        private var coreDataObserver: CoreDataObserver?
+        private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+        private var subscriptions = Set<AnyCancellable>()
 
         typealias PumpEvent = PumpEventStored.EventType
 
         override func subscribe() {
-            coreDataObserver = CoreDataObserver()
+            coreDataPublisher =
+                changedObjectsOnManagedObjectContextDidSavePublisher()
+                    .receive(on: DispatchQueue.global(qos: .background))
+                    .share()
+                    .eraseToAnyPublisher()
 
             // Parallelize Setup functions
             setupHomeViewConcurrently()
@@ -105,7 +111,7 @@ extension Home {
             Task {
                 await withTaskGroup(of: Void.self) { group in
                     group.addTask {
-                        self.setupNotification()
+                        self.registerSubscribers()
                     }
                     group.addTask {
                         self.registerHandlers()
@@ -165,43 +171,62 @@ extension Home {
             }
         }
 
+        // These combine subscribers are only necessary due to the batch inserts of glucose/FPUs which do not trigger a ManagedObjectContext change notification
+        private func registerSubscribers() {
+            glucoseStorage.updatePublisher
+                .receive(on: DispatchQueue.global(qos: .background))
+                .sink { [weak self] _ in
+                    guard let self = self else { return }
+                    self.setupGlucoseArray()
+                }
+                .store(in: &subscriptions)
+
+            carbsStorage.updatePublisher
+                .receive(on: DispatchQueue.global(qos: .background))
+                .sink { [weak self] _ in
+                    guard let self = self else { return }
+                    self.setupFPUsArray()
+                }
+                .store(in: &subscriptions)
+        }
+
         private func registerHandlers() {
-            coreDataObserver?.registerHandler(for: "OrefDetermination") { [weak self] in
+            coreDataPublisher?.filterByEntityName("OrefDetermination").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupDeterminationsArray()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+            coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupGlucoseArray()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "CarbEntryStored") { [weak self] in
+            coreDataPublisher?.filterByEntityName("CarbEntryStored").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupCarbsArray()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "PumpEventStored") { [weak self] in
+            coreDataPublisher?.filterByEntityName("PumpEventStored").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupInsulinArray()
                 self.setupLastBolus()
                 self.displayPumpStatusHighlightMessage()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "OpenAPS_Battery") { [weak self] in
+            coreDataPublisher?.filterByEntityName("OpenAPS_Battery").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupBatteryArray()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "OverrideStored") { [weak self] in
+            coreDataPublisher?.filterByEntityName("OverrideStored").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupOverrides()
-            }
+            }.store(in: &subscriptions)
 
-            coreDataObserver?.registerHandler(for: "OverrideRunStored") { [weak self] in
+            coreDataPublisher?.filterByEntityName("OverrideRunStored").sink { [weak self] _ in
                 guard let self = self else { return }
                 self.setupOverrideRunStored()
-            }
+            }.store(in: &subscriptions)
         }
 
         private func registerObservers() {
@@ -564,38 +589,6 @@ extension Home.StateModel: PumpManagerOnboardingDelegate {
     }
 }
 
-// MARK: - Setup Core Data observation
-
-extension Home.StateModel {
-    /// listens for the notifications sent when the managedObjectContext has saved!
-    func setupNotification() {
-        /// custom notification that is sent when a batch insert of glucose objects is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchInsert),
-            name: .didPerformBatchInsert,
-            object: nil
-        )
-
-        /// custom notification that is sent when a batch delete of fpus is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchDelete),
-            name: .didPerformBatchDelete,
-            object: nil
-        )
-    }
-
-    @objc private func handleBatchInsert() {
-        setupFPUsArray()
-        setupGlucoseArray()
-    }
-
-    @objc private func handleBatchDelete() {
-        setupFPUsArray()
-    }
-}
-
 // MARK: - Handle Core Data changes and update Arrays to display them in the UI
 
 extension Home.StateModel {

FreeAPS/Sources/Services/Calendar/CalendarManager.swift

@@ -19,7 +19,8 @@ final class BaseCalendarManager: CalendarManager, Injectable {
     @Injected() private var glucoseStorage: GlucoseStorage!
     @Injected() private var storage: FileStorage!
 
-    private var coreDataObserver: CoreDataObserver?
+    private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+    private var subscriptions = Set<AnyCancellable>()
 
     private var glucoseFormatter: NumberFormatter {
         let formatter = NumberFormatter()
@@ -58,12 +59,18 @@ final class BaseCalendarManager: CalendarManager, Injectable {
     init(resolver: Resolver) {
         injectServices(resolver)
         setupCurrentCalendar()
+
+        coreDataPublisher =
+            changedObjectsOnManagedObjectContextDidSavePublisher()
+                .receive(on: DispatchQueue.global(qos: .background))
+                .share()
+                .eraseToAnyPublisher()
+
         Task {
             await createEvent()
         }
-        coreDataObserver = CoreDataObserver()
         registerHandlers()
-        setupGlucoseNotification()
+        registerSubscribers()
     }
 
     let backgroundContext = CoreDataStack.shared.newTaskContext()
@@ -79,28 +86,24 @@ final class BaseCalendarManager: CalendarManager, Injectable {
     }
 
     private func registerHandlers() {
-        coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+        coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.createEvent()
             }
-        }
-    }
-
-    private func setupGlucoseNotification() {
-        /// custom notification that is sent when a batch insert of glucose objects is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchInsert),
-            name: .didPerformBatchInsert,
-            object: nil
-        )
+        }.store(in: &subscriptions)
     }
 
-    @objc private func handleBatchInsert() {
-        Task {
-            await createEvent()
-        }
+    private func registerSubscribers() {
+        glucoseStorage.updatePublisher
+            .receive(on: DispatchQueue.global(qos: .background))
+            .sink { [weak self] _ in
+                guard let self = self else { return }
+                Task {
+                    await self.createEvent()
+                }
+            }
+            .store(in: &subscriptions)
     }
 
     func requestAccessIfNeeded() async -> Bool {

FreeAPS/Sources/Services/LiveActivity/LiveActivityBridge.swift

@@ -1,4 +1,5 @@
 import ActivityKit
+import Combine
 import CoreData
 import Foundation
 import Swinject
@@ -29,6 +30,7 @@ import UIKit
     @Injected() private var settingsManager: SettingsManager!
     @Injected() private var broadcaster: Broadcaster!
     @Injected() private var storage: FileStorage!
+    @Injected() private var glucoseStorage: GlucoseStorage!
 
     private let activityAuthorizationInfo = ActivityAuthorizationInfo()
     @Published private(set) var systemEnabled: Bool
@@ -45,13 +47,20 @@ import UIKit
 
     let context = CoreDataStack.shared.newTaskContext()
 
-    private var coreDataObserver: CoreDataObserver?
+    private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+    private var subscriptions = Set<AnyCancellable>()
 
     init(resolver: Resolver) {
+        coreDataPublisher =
+            changedObjectsOnManagedObjectContextDidSavePublisher()
+                .receive(on: DispatchQueue.global(qos: .background))
+                .share()
+                .eraseToAnyPublisher()
+
         systemEnabled = activityAuthorizationInfo.areActivitiesEnabled
         injectServices(resolver)
         setupNotifications()
-        coreDataObserver = CoreDataObserver()
+        registerSubscribers()
         registerHandler()
         monitorForLiveActivityAuthorizationChanges()
         setupGlucoseArray()
@@ -59,7 +68,6 @@ import UIKit
 
     private func setupNotifications() {
         let notificationCenter = Foundation.NotificationCenter.default
-        notificationCenter.addObserver(self, selector: #selector(handleBatchInsert), name: .didPerformBatchInsert, object: nil)
         notificationCenter.addObserver(self, selector: #selector(cobOrIobDidUpdate), name: .didUpdateCobIob, object: nil)
         notificationCenter
             .addObserver(forName: UIApplication.didEnterBackgroundNotification, object: nil, queue: nil) { [weak self] _ in
@@ -73,14 +81,20 @@ import UIKit
 
     private func registerHandler() {
         // Since we are only using this info to show if an Override is active or not in the Live Activity it is enough to observe only the 'OverrideStored' Entity
-        coreDataObserver?.registerHandler(for: "OverrideStored") { [weak self] in
+        coreDataPublisher?.filterByEntityName("OverrideStored").sink { [weak self] _ in
             guard let self = self else { return }
             self.overridesDidUpdate()
-        }
+        }.store(in: &subscriptions)
     }
 
-    @objc private func handleBatchInsert() {
-        setupGlucoseArray()
+    private func registerSubscribers() {
+        glucoseStorage.updatePublisher
+            .receive(on: DispatchQueue.global(qos: .background))
+            .sink { [weak self] _ in
+                guard let self = self else { return }
+                self.setupGlucoseArray()
+            }
+            .store(in: &subscriptions)
     }
 
     @objc private func cobOrIobDidUpdate() {

FreeAPS/Sources/Services/Network/NightscoutManager.swift

@@ -72,12 +72,19 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
     private var lastEnactedDetermination: Determination?
     private var lastSuggestedDetermination: Determination?
 
-    private var coreDataObserver: CoreDataObserver?
+    private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+    private var subscriptions = Set<AnyCancellable>()
 
     init(resolver: Resolver) {
         injectServices(resolver)
         subscribe()
-        coreDataObserver = CoreDataObserver()
+
+        coreDataPublisher =
+            changedObjectsOnManagedObjectContextDidSavePublisher()
+                .receive(on: DispatchQueue.global(qos: .background))
+                .share()
+                .eraseToAnyPublisher()
+
         registerHandlers()
         setupNotification()
     }
@@ -91,42 +98,47 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
     }
 
     private func registerHandlers() {
-        coreDataObserver?.registerHandler(for: "OrefDetermination") { [weak self] in
+        coreDataPublisher?.filterByEntityName("OrefDetermination").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadStatus()
             }
-        }
-        coreDataObserver?.registerHandler(for: "OverrideStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("OverrideStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadOverrides()
             }
-        }
-        coreDataObserver?.registerHandler(for: "OverrideRunStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("OverrideRunStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadOverrides()
             }
-        }
-        coreDataObserver?.registerHandler(for: "PumpEventStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("PumpEventStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadPumpHistory()
             }
-        }
-        coreDataObserver?.registerHandler(for: "CarbEntryStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("CarbEntryStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadCarbs()
             }
-        }
-        coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task.detached {
                 await self.uploadManualGlucose()
             }
-        }
+        }.store(in: &subscriptions)
     }
 
     func setupNotification() {

FreeAPS/Sources/Services/UserNotifications/UserNotificationsManager.swift

@@ -1,4 +1,5 @@
 import AudioToolbox
+import Combine
 import CoreData
 import Foundation
 import LoopKit
@@ -56,12 +57,20 @@ final class BaseUserNotificationsManager: NSObject, UserNotificationsManager, In
     private let viewContext = CoreDataStack.shared.persistentContainer.viewContext
     private let backgroundContext = CoreDataStack.shared.newTaskContext()
 
-    private var coreDataObserver: CoreDataObserver?
+    private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+    private var subscriptions = Set<AnyCancellable>()
 
     init(resolver: Resolver) {
         super.init()
         center.delegate = self
         injectServices(resolver)
+
+        coreDataPublisher =
+            changedObjectsOnManagedObjectContextDidSavePublisher()
+                .receive(on: DispatchQueue.global(qos: .background))
+                .share()
+                .eraseToAnyPublisher()
+
         broadcaster.register(DeterminationObserver.self, observer: self)
         broadcaster.register(BolusFailureObserver.self, observer: self)
         broadcaster.register(pumpNotificationObserver.self, observer: self)
@@ -70,7 +79,7 @@ final class BaseUserNotificationsManager: NSObject, UserNotificationsManager, In
             await sendGlucoseNotification()
         }
         registerHandlers()
-        setupGlucoseNotification()
+        registerSubscribers()
         subscribeOnLoop()
     }
 
@@ -84,28 +93,24 @@ final class BaseUserNotificationsManager: NSObject, UserNotificationsManager, In
 
     private func registerHandlers() {
         // Due to the Batch insert this only is used for observing Deletion of Glucose entries
-        coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+        coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.sendGlucoseNotification()
             }
-        }
-    }
-
-    private func setupGlucoseNotification() {
-        /// custom notification that is sent when a batch insert of glucose objects is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchInsert),
-            name: .didPerformBatchInsert,
-            object: nil
-        )
+        }.store(in: &subscriptions)
     }
 
-    @objc private func handleBatchInsert() {
-        Task {
-            await sendGlucoseNotification()
-        }
+    private func registerSubscribers() {
+        glucoseStorage.updatePublisher
+            .receive(on: DispatchQueue.global(qos: .background))
+            .sink { [weak self] _ in
+                guard let self = self else { return }
+                Task {
+                    await self.sendGlucoseNotification()
+                }
+            }
+            .store(in: &subscriptions)
     }
 
     private func addAppBadge(glucose: Int?) {

FreeAPS/Sources/Services/WatchManager/WatchManager.swift

@@ -1,3 +1,4 @@
+import Combine
 import CoreData
 import Foundation
 import Swinject
@@ -17,6 +18,7 @@ final class BaseWatchManager: NSObject, WatchManager, Injectable {
     @Injected() private var carbsStorage: CarbsStorage!
     @Injected() private var tempTargetsStorage: TempTargetsStorage!
     @Injected() private var garmin: GarminManager!
+    @Injected() private var glucoseStorage: GlucoseStorage!
 
     private var glucoseFormatter: NumberFormatter {
         let formatter = NumberFormatter()
@@ -56,7 +58,8 @@ final class BaseWatchManager: NSObject, WatchManager, Injectable {
     let context = CoreDataStack.shared.newTaskContext()
     let viewContext = CoreDataStack.shared.persistentContainer.viewContext
 
-    private var coreDataObserver: CoreDataObserver?
+    private var coreDataPublisher: AnyPublisher<Set<NSManagedObject>, Never>?
+    private var subscriptions = Set<AnyCancellable>()
 
     private var lifetime = Lifetime()
 
@@ -64,9 +67,14 @@ final class BaseWatchManager: NSObject, WatchManager, Injectable {
         self.session = session
         super.init()
         injectServices(resolver)
-        setupNotification()
-        coreDataObserver = CoreDataObserver()
         registerHandlers()
+        registerSubscribers()
+
+        coreDataPublisher =
+            changedObjectsOnManagedObjectContextDidSavePublisher()
+                .receive(on: DispatchQueue.global(qos: .background))
+                .share()
+                .eraseToAnyPublisher()
 
         Task {
             await configureState()
@@ -94,42 +102,40 @@ final class BaseWatchManager: NSObject, WatchManager, Injectable {
         }
     }
 
-    func setupNotification() {
-        /// custom notification that is sent when a batch insert of glucose objects is done
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(handleBatchInsert),
-            name: .didPerformBatchInsert,
-            object: nil
-        )
-    }
-
-    @objc private func handleBatchInsert() {
-        Task {
-            await self.configureState()
-        }
+    private func registerSubscribers() {
+        glucoseStorage.updatePublisher
+            .receive(on: DispatchQueue.global(qos: .background))
+            .sink { [weak self] _ in
+                guard let self = self else { return }
+                Task {
+                    await self.configureState()
+                }
+            }
+            .store(in: &subscriptions)
     }
 
     private func registerHandlers() {
-        coreDataObserver?.registerHandler(for: "OrefDetermination") { [weak self] in
+        coreDataPublisher?.filterByEntityName("OrefDetermination").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.configureState()
             }
-        }
-        coreDataObserver?.registerHandler(for: "OverrideStored") { [weak self] in
+        }.store(in: &subscriptions)
+
+        coreDataPublisher?.filterByEntityName("OverrideStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.configureState()
             }
-        }
+        }.store(in: &subscriptions)
+
         // Observes Deletion of Glucose Objects
-        coreDataObserver?.registerHandler(for: "GlucoseStored") { [weak self] in
+        coreDataPublisher?.filterByEntityName("GlucoseStored").sink { [weak self] _ in
             guard let self = self else { return }
             Task {
                 await self.configureState()
             }
-        }
+        }.store(in: &subscriptions)
     }
 
     private func fetchlastDetermination() async -> [NSManagedObjectID] {

Model/CoreDataObserver.swift

@@ -1,46 +1,25 @@
+import Combine
 import CoreData
 import Foundation
 
-class CoreDataObserver {
-    private var entityUpdateHandlers: [String: () -> Void] = [:] // Dictionary to store pairs of entities and handlers
+func changedObjectsOnManagedObjectContextDidSavePublisher() -> some Publisher<Set<NSManagedObject>, Never> {
+    Foundation.NotificationCenter.default
+        .publisher(for: NSNotification.Name.NSManagedObjectContextDidSave)
+        .map { notification in
+            guard let userInfo = notification.userInfo else { return Set<NSManagedObject>() }
 
-    init() {
-        setupNotification()
-    }
-
-    func registerHandler(for entityName: String, handler: @escaping () -> Void) {
-        entityUpdateHandlers[entityName] = handler
-    }
-
-    private func setupNotification() {
-        Foundation.NotificationCenter.default.addObserver(
-            self,
-            selector: #selector(contextDidSave(_:)),
-            name: NSNotification.Name.NSManagedObjectContextDidSave,
-            object: nil
-        )
-    }
+            var objects = Set((userInfo[NSInsertedObjectsKey] as? Set<NSManagedObject>) ?? [])
+            objects.formUnion((userInfo[NSUpdatedObjectsKey] as? Set<NSManagedObject>) ?? [])
+            objects.formUnion((userInfo[NSDeletedObjectsKey] as? Set<NSManagedObject>) ?? [])
 
-    @objc private func contextDidSave(_ notification: Notification) {
-        guard let userInfo = notification.userInfo else { return }
-
-        Task {
-            await processUpdates(userInfo: userInfo)
+            return objects
         }
-    }
-
-    private func processUpdates(userInfo: [AnyHashable: Any]) async {
-        var objects = Set((userInfo[NSInsertedObjectsKey] as? Set<NSManagedObject>) ?? [])
-        objects.formUnion((userInfo[NSUpdatedObjectsKey] as? Set<NSManagedObject>) ?? [])
-        objects.formUnion((userInfo[NSDeletedObjectsKey] as? Set<NSManagedObject>) ?? [])
+}
 
-        for (entityName, handler) in entityUpdateHandlers {
-            let entityUpdates = objects.filter { $0.entity.name == entityName }
-            DispatchQueue.global(qos: .background).async {
-                if entityUpdates.isNotEmpty {
-                    handler()
-                }
-            }
+extension Publisher where Output == Set<NSManagedObject> {
+    func filterByEntityName(_ name: String) -> some Publisher<Self.Output, Self.Failure> {
+        filter { objects in
+            objects.contains(where: { $0.entity.name == name })
         }
     }
 }

Model/Helper/CustomNotification.swift

@@ -1,10 +1,6 @@
 import Foundation
 
 extension Notification.Name {
-    static let didPerformBatchInsert = Notification.Name("didPerformBatchInsert")
-    static let didPerformBatchUpdate = Notification.Name("didPerformBatchUpdate")
-    static let didPerformBatchDelete = Notification.Name("didPerformBatchDelete")
-    static let didUpdateDetermination = Notification.Name("didUpdateDetermination")
     static let didUpdateOverrideConfiguration = Notification.Name("didUpdateOverrideConfiguration")
     static let didUpdateCobIob = Notification.Name("didUpdateCobIob")
 }