How to: Using Combine to react to CoreData changes in the background

Question

I want to achieve the following: Whenever someone triggers a CoreData save (ie. NSManagedObjectContextDidSave notification gets sent), I'd like to perform some background calculation based the changed NSManagedObject. Concrete example: Assume in a notes app, I want to asynchronously calculate the total number of words in all notes.

The problem currently lies with the fact that NSManagedObject context is explicitly bound to thread and you are discouraged from using NSManagedObjects outside this thread.

I have setup two NSManagedObjectContexts in my SceneDelegate:

let context = (UIApplication.shared.delegate as! AppDelegate).persistentContainer.viewContext
let backgroundContext = (UIApplication.shared.delegate as! AppDelegate).persistentContainer.newBackgroundContext()

I also have subscribed to the notification via NotificationCenter.default.publisher(for: .NSManagedObjectContextDidSave) and am receiving a save notification twice after I trigger only one managedObjectContext.save(). However, both notifications are sent from the same thread (which is the UIThread) and all NSManagedObjects in the user dictionary have a .managedObjectContext which is the viewContext and not the backgroundContext.

My idea was to filter the notifications based on whether or not the associated NSManagedObjectContext was the background one as I assumed that the notification is also sent on the (private) DispatchQueue but it seems all notifications are sent on the UIThread and the background context is never used.

Any idea on how to solve this? Is this a bug? How can I retrieve notifications based on the backgroundContext with downstream tasks being run on the associated DispatchQueue?

Answer 1

You can create a Publisher which informs you when something relevant for you in Core Data has changed.

I wrote an article on this. Combine, Publishers and Core Data.

import Combine
import CoreData
import Foundation

class CDPublisher<Entity>: NSObject, NSFetchedResultsControllerDelegate, Publisher where Entity: NSManagedObject {
    typealias Output = [Entity]
    typealias Failure = Error

    private let request: NSFetchRequest<Entity>
    private let context: NSManagedObjectContext
    private let subject: CurrentValueSubject<[Entity], Failure>
    private var resultController: NSFetchedResultsController<NSManagedObject>?
    private var subscriptions = 0

      init(request: NSFetchRequest<Entity>, context: NSManagedObjectContext) {
        if request.sortDescriptors == nil { request.sortDescriptors = [] }
        self.request = request
        self.context = context
        subject = CurrentValueSubject([])
        super.init()
    }

      func receive<S>(subscriber: S)
        where S: Subscriber, CDPublisher.Failure == S.Failure, CDPublisher.Output == S.Input {
        var start = false

        synchronized(self) {
            subscriptions += 1
            start = subscriptions == 1
        }

        if start {
            let controller = NSFetchedResultsController(fetchRequest: request, managedObjectContext: context, 
                                                        sectionNameKeyPath: nil, cacheName: nil)
            controller.delegate = self

            do {
                try controller.performFetch()
                let result = controller.fetchedObjects ?? []
                subject.send(result)
            } catch {
                subject.send(completion: .failure(error))
            }
            resultController = controller as? NSFetchedResultsController<NSManagedObject>
        }
        CDSubscription(fetchPublisher: self, subscriber: AnySubscriber(subscriber))
    }

      func controllerDidChangeContent(_ controller: NSFetchedResultsController<NSFetchRequestResult>) {
        let result = controller.fetchedObjects as? [Entity] ?? []
        subject.send(result)
    }

      private func dropSubscription() {
        objc_sync_enter(self)
        subscriptions -= 1
        let stop = subscriptions == 0
        objc_sync_exit(self)

        if stop {
            resultController?.delegate = nil
            resultController = nil
        }
    }

    private class CDSubscription: Subscription {
        private var fetchPublisher: CDPublisher?
        private var cancellable: AnyCancellable?

        @discardableResult
        init(fetchPublisher: CDPublisher, subscriber: AnySubscriber<Output, Failure>) {
            self.fetchPublisher = fetchPublisher

            subscriber.receive(subscription: self)

            cancellable = fetchPublisher.subject.sink(receiveCompletion: { completion in
                subscriber.receive(completion: completion)
            }, receiveValue: { value in
                _ = subscriber.receive(value)
            })
        }

        func request(_ demand: Subscribers.Demand) {}

        func cancel() {
            cancellable?.cancel()
            cancellable = nil
            fetchPublisher?.dropSubscription()
            fetchPublisher = nil
        }
    }

}