reconcileAndLocalSaveOperationSink get overwritten

[ruiguoamz]

Describe the bug
In AWSModelReconciliationQueue class:

func enqueue(_ remoteModel: MutationSync<AnyModel>) {
    let reconcileOp = ReconcileAndLocalSaveOperation(remoteModel: remoteModel,
                                                     storageAdapter: storageAdapter)
    reconcileAndLocalSaveOperationSink = reconcileOp.publisher.sink(receiveCompletion: { error in
        self.modelReconciliationQueueSubject.send(completion: error)
    }, receiveValue: { mutationEvent in
        self.modelReconciliationQueueSubject.send(.mutationEvent(mutationEvent))
    })
    reconcileAndSaveQueue.addOperation(reconcileOp)
}

whenever comes a remoteModel, it creates a reconcileOp and subscribes to the reconcileOp.publisher. But reconcileAndLocalSaveOperationSink will keep being replaced by newer reconcileOp.publisher.sink() and cause an issue to receive only one mutationEvent.

Related customer issue: #598

Expected behavior

• There should be a reconcileAndLocalSaveOperationSink that is able to receive every mutationEvent emitted by reconcileAndLocalSaveOperation.publisher
• Or have a reconcileAndLocalSaveOperationSink for each reconcileAndLocalSaveOperation.

[lawmicha]

reproduce this with unit test that will fail, and then apply fix to fix unit test

[lawmicha]

func enqueue(_ remoteModel: MutationSync<AnyModel>) {
        let reconcileOp = ReconcileAndLocalSaveOperation(remoteModel: remoteModel,
                                                         storageAdapter: storageAdapter)
        var reconcileAndLocalSaveOperationSink: AnyCancellable?
        reconcileAndLocalSaveOperationSink = reconcileOp.publisher.sink(receiveCompletion: { error in
            self.modelReconciliationQueueSubject.send(completion: error)
            self.reconcileAndLocalSaveOperationSinks.remove(reconcileAndLocalSaveOperationSink)
        }, receiveValue: { mutationEvent in
            self.modelReconciliationQueueSubject.send(.mutationEvent(mutationEvent))
        })
        reconcileAndLocalSaveOperationSinks.insert(reconcileAndLocalSaveOperationSink)
        reconcileAndSaveQueue.addOperation(reconcileOp)
    }