Часто в мобильном приложении необходимо реализовать следующий функционал:

Выполнить асинхронный запрос
Забиндить результат в главном потоке на различные view
Если нужно, то асинхронно обновить базу данных на устройстве в фоновом потоке
Если возникают ошибки при выполнении этих операций, то показать уведомление
Соблюсти принцип SSOT для актуальности данных
Всё это протестировать

Решить эту задачу сильно упрощает архитектурный подход MVVM и фреймворки RxSwift, CoreData.

Описанный ниже подход использует принципы реактивного программирования и не привязан исключительно к RxSwift и CoreData. И при желании может быть реализован с помощью других инструментов.

В качестве примера я возьму фрагмент приложения в котором отображаются данные продавца. В контроллере два аутлета UILabel для телефона и адреса и одна UIButton для звонка по этому телефону. ContactsViewController.

Объясню реализацию от model к view.

Model

Фрагмент автосгенерированного файла SellerContacts+CoreDataProperties из DerivedSources
с атрибутами:

extension SellerContacts {

    @nonobjc public class func fetchRequest() -> NSFetchRequest<SellerContacts> {
        return NSFetchRequest<SellerContacts>(entityName: "SellerContacts")
    }

    @NSManaged public var address: String?
    @NSManaged public var order: Int16
    @NSManaged public var phone: String?

}

Repository.

Метод предоставляющий данные продавца:

func sellerContacts() -> Observable<Event<[SellerContacts]>> {
        // 1
        Observable.merge([
            // 2
            container.viewContext.rx.entities(fetchRequest: SellerContacts.fetchRequestWithSort()).materialize(),
            // 3
            updater.sync()
        ])
    }

Как раз в этом месте реализуется SSOT. Запрос делается к CoreData, и CoreData обновляется, если необходимо. Все данные получаются ТОЛЬКО из БД, а updater.sync() может сгенерировать только Event с ошибкой, но НЕ с данными.

Использование оператора merge позволяет нам добиться асинхронности выполнения запроса к базе данных и её обновления.
Для удобства построения запроса к БД используется RxCoreData
Выполняем обновление БД

Т.к. используется асинхронный подход получения и обновления данных, необходимо использовать Observable<Event<...>>. Это нужно для того, чтобы subscriber не получил Error, при ошибке во время получения remote data, а только показал эту ошибку и продолжал реагировать на изменения в CoreData. Об этом подробнее чуть позже.

DatabaseUpdater
В приложении из примера удаленные данные получаются из Firebase Remote Config. CoreData обновляется только в том случае, если fetchAndActivate() завершается со статусом .successFetchedFromRemote.

Но можно использовать любые другие ограничения обновления, например, по времени.
Метод sync() для обновления БД:

func sync<T>() -> Observable<Event<T>> {
        // 1
        // Check can fetch
        if fetchLimiter.fetchInProcess {
            return Observable.empty()
        }
        // 2
        // Block fetch for other requests
        fetchLimiter.fetchInProcess = true
        // 3
        // Fetch & activate remote config
        return remoteConfig.rx.fetchAndActivate().flatMap { [weak self] status, error -> Observable<Event<T>> in
            // 4
            // Default result
            var result = Observable<Event<T>>.empty()
            // Update database only when config wethed from remote
            switch status {
            // 5
            case .error:
                let error = error ?? AppError.unknown
                print("Remote config fetch error: \(error.localizedDescription)")
                // Set error to result
                result = Observable.just(Event.error(error))
            // 6
            case .successFetchedFromRemote:
                print("Remote config fetched data from remote")
                // Update database from remote config
                result = self?.update() ?? Observable.empty()
            case .successUsingPreFetchedData:
                print("Remote config using prefetched data")
            @unknown default:
                print("Remote config unknown status")
            }
            // 7
            // Unblock fetch for other requests
            self?.fetchLimiter.fetchInProcess = false
            return result
        }
    }

Возвращаем пустую последовательность, если получение данных уже идет. Например, другой метод из репозитория уже вызвал sync(). fetchLimiter должен быть потокобезопасным. А именно, получать или записывать значения в поле fetchInProcess нужно в последовательной очереди.
Блокируем обновление для последующих вызовов метода
Выполняем запрос для получения удаленных данных
Создаем результат с пустой последовательностью по умолчанию
Если запрос выполнился с ошибкой то присваиваем результату последовательность с одним элементом Event с ошибкой
Обновляем БД
Включаем возможность обновления БД и возвращаем результат

Multithreading

Для обеспечения многопоточности получаем данные из Core Data viewContext в главном потоке, а обновляем данные при помощи метода

func performBackgroundTask(_ block: @escaping (NSManagedObjectContext) -> Void)

из NSPersistentContainer.
И не забываем про

container.viewContext.automaticallyMergesChangesFromParent = true

во время создания контейнера.

/** Update database from remote data */
    private func update<T>() -> Observable<Event<T>> {
        Observable.create { [weak self] observer in
            self?.container.performBackgroundTask { [weak self] context in
                do {
                    ...
                    try self?.updateSellerContacts(context)
                    ...
                    if context.hasChanges {
                        try context.save()
                    }
                    observer.onCompleted()
                } catch {
                    context.undo()
                    observer.onNext(Event.error(error))
                    observer.onCompleted()
                }
            }
            return Disposables.create()
        }.subscribeOn(CurrentThreadScheduler.instance)
    }

ViewModel
В данном примере во ViewModel просто вызывается метод sellerContacts() из Repository и возвращается результат.

func contacts() -> Observable<Event<[SellerContacts]>> {
        repository.sellerContacts()
    }

ViewController
В контроллере нужно забиндить результат запроса в поля. Для этого в viewDidLoad() вызывается метод bindContacts():

private func bindContacts() {
        // 1
        viewModel?.contacts()
            .observeOn(MainScheduler.instance)
             // 2
            .flatMapError { [weak self] in
                self?.rx.showMessage($0.localizedDescription) ?? Observable.empty()
            }
             // 3
            .compactMap { $0.first }
             // 4
            .subscribe(onNext: { [weak self] in
                self?.phone.text = $0.phone
                self?.address.text = $0.address
            }).disposed(by: disposeBag)
    }

Выполняем запрос контактов. Явно указываем главный поток для работы с полученным результатом, т.к. данные будут устанавливаться во view
Если приходит элемент содержащий Event с ошибкой, то показывается сообщение с ошибкой и возвращается пустая последовательность. Подробнее об операторе flatMapError и showMessage ниже
Используем оператор compactMap для получения контактов из массива
Устанавливаем данные в аутлеты

Оператор .flatMapError()
Для преобразования результата последовательности из Event в элемент в нём содержащийся или показа ошибки используется оператор:

func flatMapError<T>(_ handler: ((_ error: Error) -> Observable<T>)? = nil) -> Observable<Element.Element> {
        // 1
        flatMap { element -> Observable<Element.Element> in
            switch element.event {
            // 2
            case .error(let error):
                return handler?(error).flatMap { _ in Observable<Element.Element>.empty() } ?? Observable.empty()
            // 3
            case .next(let element):
                return Observable.just(element)
            // 4
            default:
                return Observable.empty()
            }
        }
    }

Преобразуем последовательность из Event.Element в Element
Если Event содержит ошибку, то возвращаем handler преобразованный в пустую последовательность
Если Event содержит результат, то возвращаем последовательность с одним элементом, содержащим этот результат
По умолчанию возвращается пустая последовательность

Такой подход позволяет обрабатывать ошибки выполнения запросов, не посылая подписчику Error Event. И наблюдение за изменением в БД остаётся активным.

Оператор .showMessage()
Для показа сообщений пользователю используется оператор:

public func showMessage(_ text: String, withEvent: Bool = false) -> Observable<Void> {
        // 1
        let _alert = alert(title: nil,
              message: text,
              actions: [AlertAction(title: "OK", style: .default)]
        // 2
        ).map { _ in () }
        // 3
        return withEvent ? _alert : _alert.flatMap { Observable.empty() }
    }

С помощью RxAlert создаётся окно с сообщением и одной кнопкой
Результат преобразуется в Void
Если необходимо событие после показа сообщения, то возвращаем результат. Иначе сначала преобразуем его в пустую последовательность, а затем возвращаем

Т.к. .showMessage() может использоваться не только для показа уведомлений об ошибках, то полезно иметь возможность регулировать какая последовательность получается в итоге — пустая или с событием.

Тесты

Все описанное выше протестировать не трудно. Начнем по порядку изложения.

RepositoryTests
Для теста репозитория используется DatabaseUpdaterMock. Там есть возможность отслеживать вызывался ли метод sync() и устанавливать результат его выполнения:

func testSellerContacts() throws {
        // 1
        // Success
        // Check sequence contains only one element
        XCTAssertThrowsError(try repository.sellerContacts().take(2).toBlocking(timeout: 1).toArray())
        updater.isSync = false
        // Check that element
        var result = try repository.sellerContacts().toBlocking().first()?.element
        XCTAssertTrue(updater.isSync)
        XCTAssertEqual(result?.count, sellerContacts.count)

        // 2
        // Sync error
        updater.isSync = false
        updater.error = AppError.unknown
        let resultArray = try repository.sellerContacts().take(2).toBlocking().toArray()
        XCTAssertTrue(resultArray.contains { $0.error?.localizedDescription == AppError.unknown.localizedDescription })
        XCTAssertTrue(updater.isSync)
        result = resultArray.first { $0.error == nil }?.element
        XCTAssertEqual(result?.count, sellerContacts.count)
    }

Проверяем, что последовательность содержит только один элемент, вызывается метод sync()
Проверяем, что последовательность содержит два элемента. Один содержит Event с ошибкой, другой результат запроса из БД, вызывается метод sync()

DatabaseUpdaterTests

testSync()

func testSync() throws {
        let remoteConfig = RemoteConfigMock()
        let fetchLimiter = FetchLimiter(serialQueue: DispatchQueue(label: "test"))
        let databaseUpdater = DatabaseUpdaterImpl(remoteConfig: remoteConfig, decoder: JSONDecoderMock(), context: context, fetchLimiter: fetchLimiter)
        // 1
        // Not update. Fetch in process
        fetchLimiter.fetchInProcess = true
        XCTAssertFalse(remoteConfig.isFetchAndActivate)
        XCTAssertFalse(remoteConfig.isSubscript)
        
        expectation(forNotification: .NSManagedObjectContextDidSave, object: context)
            .isInverted = true
    
        var sync: Observable<Event<Void>> = databaseUpdater.sync()
        XCTAssertNil(try sync.toBlocking().first())
        XCTAssertFalse(remoteConfig.isFetchAndActivate)
        XCTAssertFalse(remoteConfig.isSubscript)
        XCTAssertTrue(fetchLimiter.fetchInProcess)
        
        waitForExpectations(timeout: 1)
        // 2
        // Not update. successUsingPreFetchedData
        fetchLimiter.fetchInProcess = false
        
        expectation(forNotification: .NSManagedObjectContextDidSave, object: context)
            .isInverted = true
        
        sync = databaseUpdater.sync()
        var result: Event<Void>?
        sync.subscribe(onNext: { result = $0 }).disposed(by: disposeBag)
        XCTAssertTrue(fetchLimiter.fetchInProcess)
        remoteConfig.completionHandler?(RemoteConfigFetchAndActivateStatus.successUsingPreFetchedData, nil)
        
        waitForExpectations(timeout: 1)
        XCTAssertNil(result)
        XCTAssertTrue(remoteConfig.isFetchAndActivate)
        XCTAssertFalse(remoteConfig.isSubscript)
        XCTAssertFalse(fetchLimiter.fetchInProcess)
        // 3
        // Not update. Error
        fetchLimiter.fetchInProcess = false
        remoteConfig.isFetchAndActivate = false
        
        expectation(forNotification: .NSManagedObjectContextDidSave, object: context)
            .isInverted = true
        sync = databaseUpdater.sync()
        sync.subscribe(onNext: { result = $0 }).disposed(by: disposeBag)
        XCTAssertTrue(fetchLimiter.fetchInProcess)
        remoteConfig.completionHandler?(RemoteConfigFetchAndActivateStatus.error, AppError.unknown)
        
        waitForExpectations(timeout: 1)
        
        XCTAssertEqual(result?.error?.localizedDescription, AppError.unknown.localizedDescription)
        XCTAssertTrue(remoteConfig.isFetchAndActivate)
        XCTAssertFalse(remoteConfig.isSubscript)
        XCTAssertFalse(fetchLimiter.fetchInProcess)
        // 4
        // Update
        fetchLimiter.fetchInProcess = false
        remoteConfig.isFetchAndActivate = false
        result = nil
        
        expectation(forNotification: .NSManagedObjectContextDidSave, object: context)
        
        sync = databaseUpdater.sync()
        sync.subscribe(onNext: { result = $0 }).disposed(by: disposeBag)
        XCTAssertTrue(fetchLimiter.fetchInProcess)
        remoteConfig.completionHandler?(RemoteConfigFetchAndActivateStatus.successFetchedFromRemote, nil)
        
        waitForExpectations(timeout: 1)
        
        XCTAssertNil(result)
        XCTAssertTrue(remoteConfig.isFetchAndActivate)
        XCTAssertTrue(remoteConfig.isSubscript)
        XCTAssertFalse(fetchLimiter.fetchInProcess)
    }

Возвращается пустая последовательность, если обновление в процессе
Возвращается пустая последовательность, если данные не получены
Возвращается Event с ошибкой
Возвращается пустая последовательность, если данные обновились

ViewModelTests

ViewControllerTests

testBindContacts()

func testBindContacts() {
        // 1
        // Error. Show message
        XCTAssertNotEqual(controller.phone.text, contacts.phone)
        XCTAssertNotEqual(controller.address.text, contacts.address)
        viewModel.contactsResult.accept(Event.error(AppError.unknown))
        
        expectation(description: "wait 1 second").isInverted = true
        waitForExpectations(timeout: 1)
        // 2
        XCTAssertNotNil(controller.presentedViewController)
        let alertController = controller.presentedViewController as! UIAlertController
        XCTAssertEqual(alertController.actions.count, 1)
        XCTAssertEqual(alertController.actions.first?.style, .default)
        XCTAssertEqual(alertController.actions.first?.title, "OK")
        XCTAssertNotEqual(controller.phone.text, contacts.phone)
        XCTAssertNotEqual(controller.address.text, contacts.address)
        // 3
        // Trigger action OK
        let action = alertController.actions.first!
        typealias AlertHandler = @convention(block) (UIAlertAction) -> Void
        let block = action.value(forKey: "handler")
        let blockPtr = UnsafeRawPointer(Unmanaged<AnyObject>.passUnretained(block as AnyObject).toOpaque())
        let handler = unsafeBitCast(blockPtr, to: AlertHandler.self)
        handler(action)
        
        expectation(description: "wait 1 second").isInverted = true
        waitForExpectations(timeout: 1)
        // 4
        XCTAssertNil(controller.presentedViewController)
        XCTAssertNotEqual(controller.phone.text, contacts.phone)
        XCTAssertNotEqual(controller.address.text, contacts.address)
        // 5
        // Empty array of contats
        viewModel.contactsResult.accept(Event.next([]))
        
        expectation(description: "wait 1 second").isInverted = true
        waitForExpectations(timeout: 1)
        
        XCTAssertNil(controller.presentedViewController)
        XCTAssertNotEqual(controller.phone.text, contacts.phone)
        XCTAssertNotEqual(controller.address.text, contacts.address)
        // 6
        // Success
        viewModel.contactsResult.accept(Event.next([contacts]))
        
        expectation(description: "wait 1 second").isInverted = true
        waitForExpectations(timeout: 1)
        
        XCTAssertNil(controller.presentedViewController)
        XCTAssertEqual(controller.phone.text, contacts.phone)
        XCTAssertEqual(controller.address.text, contacts.address)
    }

Показать сообщение об ошибке
Проверить, что в controller.presentedViewController сообщение об ошибке
Выполнить handler для кнопки Ок и убедиться, что окно с сообщением скрылось
Для пустого результата не показывается ошибка и не заполняются поля
Для успешного запроса не показывается ошибка и заполняются поля

Тесты для операторов

.flatMapError()
.showMessage()

Используя подобный подход в проектировании мы реализуем асинхронное получение, обновление данных и уведомление об ошибках без потери возможности реагировать на изменение данных, следуя принципу SSOT.

Single source of truth (SSOT) on MVVM with RxSwift & CoreData