added mechanism to track transaction sources

Sources/ListPublisher.swift

@@ -76,13 +76,7 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
     /**
      A snapshot of the latest state of this list
      */
-    public fileprivate(set) var snapshot: ListSnapshot<O> = .init() {
-
-        didSet {
-
-            self.notifyObservers()
-        }
-    }
+    public private(set) var snapshot: ListSnapshot<O> = .init()
 
 
     // MARK: Public (Observers)
@@ -111,7 +105,7 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
             "Attempted to add an observer of type \(Internals.typeName(observer)) outside the main thread."
         )
         self.observers.setObject(
-            Internals.Closure(callback),
+            Internals.Closure({ callback($0.listPublisher) }),
             forKey: observer
         )
         if notifyInitial {
@@ -119,6 +113,44 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
             callback(self)
         }
     }
+    
+    /**
+     Registers an object as an observer to be notified when changes to the `ListPublisher`'s snapshot occur.
+
+     To prevent retain-cycles, `ListPublisher` only keeps `weak` references to its observers.
+
+     For thread safety, this method needs to be called from the main thread. An assertion failure will occur (on debug builds only) if called from any thread other than the main thread.
+
+     Calling `addObserver(_:_:)` multiple times on the same observer is safe.
+
+     - parameter observer: an object to become owner of the specified `callback`
+     - parameter notifyInitial: if `true`, the callback is executed immediately with the current publisher state. Otherwise only succeeding updates will notify the observer. Default value is `false`.
+     - parameter initialSourceIdentifier: an optional value that identifies the initial callback invocation if `notifyInitial` is `true`.
+     - parameter callback: the closure to execute when changes occur
+     */
+    public func addObserver<T: AnyObject>(
+        _ observer: T,
+        notifyInitial: Bool = false,
+        initialSourceIdentifier: Any? = nil,
+        _ callback: @escaping (
+            _ listPublisher: ListPublisher<O>,
+            _ sourceIdentifier: Any?
+        ) -> Void
+    ) {
+
+        Internals.assert(
+            Thread.isMainThread,
+            "Attempted to add an observer of type \(Internals.typeName(observer)) outside the main thread."
+        )
+        self.observers.setObject(
+            Internals.Closure(callback),
+            forKey: observer
+        )
+        if notifyInitial {
+            
+            callback(self, initialSourceIdentifier)
+        }
+    }
 
     /**
      Unregisters an object from receiving notifications for changes to the `ListPublisher`'s snapshot.
@@ -301,6 +333,20 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
     }
     
     
+    // MARK: FilePrivate
+    
+    fileprivate typealias ObserverClosureType = Internals.Closure<(listPublisher: ListPublisher<O>, sourceIdentifier: Any?), Void>
+    
+    fileprivate func set(
+        snapshot: ListSnapshot<O>,
+        sourceIdentifier: Any?
+    )  {
+
+        self.snapshot = snapshot
+        self.notifyObservers(sourceIdentifier: sourceIdentifier)
+    }
+    
+    
     // MARK: Private
 
     private var query: (
@@ -315,7 +361,7 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
     private var observerForWillChangePersistentStore: Internals.NotificationObserver!
     private var observerForDidChangePersistentStore: Internals.NotificationObserver!
 
-    private lazy var observers: NSMapTable<AnyObject, Internals.Closure<ListPublisher<O>, Void>> = .weakToStrongObjects()
+    private lazy var observers: NSMapTable<AnyObject, ObserverClosureType> = .weakToStrongObjects()
 
     private static func recreateFetchedResultsController(context: NSManagedObjectContext, from: From<ObjectType>, sectionBy: SectionBy<ObjectType>?, applyFetchClauses: @escaping (_ fetchRequest: Internals.CoreStoreFetchRequest<NSManagedObject>) -> Void) -> (controller: Internals.CoreStoreFetchedResultsController, delegate: Internals.FetchedDiffableDataSourceSnapshotDelegate) {
 
@@ -359,15 +405,19 @@ public final class ListPublisher<O: DynamicObject>: Hashable {
         try! self.fetchedResultsController.performFetchFromSpecifiedStores()
     }
 
-    private func notifyObservers() {
+    private func notifyObservers(sourceIdentifier: Any?) {
 
         guard let enumerator = self.observers.objectEnumerator() else {
 
             return
         }
+        let arguments: ObserverClosureType.Arguments = (
+            listPublisher: self,
+            sourceIdentifier: sourceIdentifier
+        )
         for closure in enumerator {
 
-            (closure as! Internals.Closure<ListPublisher<O>, Void>).invoke(with: self)
+            (closure as! ObserverClosureType).invoke(with: arguments)
         }
     }
 }
@@ -384,11 +434,18 @@ extension ListPublisher: FetchedDiffableDataSourceSnapshotHandler {
         return self.query.sectionIndexTransformer
     }
 
-    internal func controller(_ controller: NSFetchedResultsController<NSFetchRequestResult>, didChangeContentWith snapshot: Internals.DiffableDataSourceSnapshot) {
+    internal func controller(
+        _ controller: NSFetchedResultsController<NSFetchRequestResult>,
+        didChangeContentWith snapshot: Internals.DiffableDataSourceSnapshot
+    ) {
 
-        self.snapshot = .init(
-            diffableSnapshot: snapshot,
-            context: controller.managedObjectContext
+        let context = controller.managedObjectContext
+        self.set(
+            snapshot: .init(
+                diffableSnapshot: snapshot,
+                context: context
+            ),
+            sourceIdentifier: context.saveMetadata?.sourceIdentifier
         )
     }
 }