Fix: Properly keep the cancellables

DVDRental was stuck in connecting. This was because
the Combine streams were not held onto. Originally
I thought the streams could "self retain", but that
doesn't fly anymore.

So added tokens for
- count query
- range query
- model fetch

And DVDRental seems to be back in a working state.

Author: helje5

Sources/DirectToSwiftUI/ViewModel/D2SDisplayGroup.swift

@@ -120,7 +120,8 @@ public final class D2SDisplayGroup<Object: OActiveRecord>
 
   private func integrateCount(_ count: Int) {
     assert(_dispatchPreconditionTest(.onQueue(.main)))
-    
+    self.countFetchToken = nil
+
     #if false // nope, a fetch count means we rebuild!
       if count == results.count { return } // all good already
     #endif
@@ -138,11 +139,14 @@ public final class D2SDisplayGroup<Object: OActiveRecord>
     results.clearOrderAndApplyNewCount(count)
   }
 
+  private var countFetchToken : AnyCancellable?
+  
   private func fetchCount(_ fetchSpecification: FetchSpecification) {
     let fs = fetchSpecification // has to be done, can't use inside fetchCount?
-    _ = dataSource.fetchCount(fs, on: D2SFetchQueue)
+    countFetchToken = dataSource.fetchCount(fs, on: D2SFetchQueue)
       .receive(on: RunLoop.main)
       .catch { ( error : Swift.Error ) -> Just<Int> in
+        self.countFetchToken = nil
         self.handleError(error)
         return Just(0)
       }
@@ -196,8 +200,13 @@ public final class D2SDisplayGroup<Object: OActiveRecord>
     self.results = newResults
   }
 
-  private struct Query: Equatable {
+  private final class Query: Equatable {
     let range : Range<Int>
+    var token : AnyCancellable?
+    init(range: Range<Int>) { self.range = range }
+    static func ==(lhs: Query, rhs: Query) -> Bool {
+      return lhs.range == rhs.range
+    }
   }
   private var activeQueries = [ Query ]()
 
@@ -242,7 +251,7 @@ public final class D2SDisplayGroup<Object: OActiveRecord>
     let query = Query(range: fetchRange)
     activeQueries.append(query) // keep it alive
 
-    _ = dataSource.fetchGlobalIDs(fs, on: D2SFetchQueue)
+    query.token = dataSource.fetchGlobalIDs(fs, on: D2SFetchQueue)
       .receive(on: RunLoop.main)
       .flatMap { ( globalIDs ) -> AnyPublisher<[ Object ], Error> in
         var missingGIDs = Set<GlobalID>()

Sources/DirectToSwiftUI/ViewModel/D2SRuleEnvironment.swift

@@ -43,18 +43,21 @@ public final class D2SRuleEnvironment: ObservableObject {
     ruleContext[D2SKeys.model] = model
   }
 
+  private var modelFetch : AnyCancellable?
+  
   public func resume() {
     guard databaseModel == nil else { return }
 
     // TODO: setup timer to refetch and compare model tag
-    _ = adaptor.fetchModel(on: D2SFetchQueue)
+    modelFetch = adaptor.fetchModel(on: D2SFetchQueue)
       .map { ( model, tag ) in
         FancyModelMaker(model: model).fancyfyModel()
       }
       .receive(on: RunLoop.main)
       .catch { ( error : Swift.Error ) -> Just<Model> in
         self.error = error
         globalD2SLogger.error("failed to fetch model:", error)
+        self.modelFetch = nil
         return Just(Model(entities: [
           ModelEntity(name: "Could not load model.")
         ]))
@@ -64,6 +67,7 @@ public final class D2SRuleEnvironment: ObservableObject {
           self.adaptor.model = model
         }
         self.setupWithModel(model)
+        self.modelFetch = nil
       }
   }
 }