Swift Combine: Subsequent Publisher That Consumes Other Publishers (Using Combinelatest) Doesn't "Fire"

Swift Combine: subsequent Publisher that consumes other Publishers (using CombineLatest) doesn't fire

I've got this question answered here: https://forums.swift.org/t/crash-in-swiftui-app-using-combine-was-using-published-in-conjunction-with-state-in-swiftui/26628/9 by the very friendly and helpful Nanu Jogi, who is not on stackoverflow.

It is rather straight forward:

add this line:

        .receive(on: RunLoop.main) // run on main thread

in validatedCredentials so that it looks like this:

var validatedCredentials: AnyPublisher<(String, String)?, Never> {
    return Publishers.CombineLatest(validatedEMail, validatedPassword)

        .receive(on: RunLoop.main) // <<—— run on main thread

        .map { validatedEMail, validatedPassword in
            print("validatedEMail: \(validatedEMail ?? "not set"), validatedPassword: \(validatedPassword ?? "not set")")

            guard let eMail = validatedEMail, let password = validatedPassword else { return nil }

            return (eMail, password)

    }
    .eraseToAnyPublisher()

This is all what is needed.

And here one more time the whole code for reference (updated for Xcode 11.0 beta 5 (11M382q)):

//
//  RegistrationView.swift
//  Combine-Beta-Feedback
//
//  Created by Lars Sonchocky-Helldorf on 09.07.19.
//  Copyright © 2019 Lars Sonchocky-Helldorf. All rights reserved.
//

import SwiftUI
import Combine

struct RegistrationView : View {
    @ObservedObject var registrationModel = RegistrationModel()

    @State private var registrationButtonDisabled = true

    @State private var validatedEMail: String = ""
    @State private var validatedPassword: String = ""

    var body: some View {
        Form {
            Section {
                TextField("Enter your EMail", text: $registrationModel.eMail)
                SecureField("Enter a Password", text: $registrationModel.password)
                SecureField("Enter the Password again", text: $registrationModel.passwordRepeat)
                Button(action: registrationButtonAction) {
                    Text("Create Account")
                }
                .disabled($registrationButtonDisabled.wrappedValue)
                    .onReceive(self.registrationModel.validatedCredentials) { newValidatedCredentials in
                        self.registrationButtonDisabled = (newValidatedCredentials == nil)
                }
            }

            Section {
                Text("Validated EMail: \(validatedEMail)")
                    .onReceive(self.registrationModel.validatedEMail) { newValidatedEMail in
                        self.validatedEMail = newValidatedEMail != nil ? newValidatedEMail! : "EMail invalid"
                }
                Text("Validated Password: \(validatedPassword)")
                    .onReceive(self.registrationModel.validatedPassword) { newValidatedPassword in
                        self.validatedPassword = newValidatedPassword != nil ? newValidatedPassword! : "Passwords to short or don't match"
                }
            }
        }
        .navigationBarTitle(Text("Sign Up"))
    }

    func registrationButtonAction() {

    }
}

class RegistrationModel : ObservableObject {

    @Published var eMail: String = ""
    @Published var password: String = ""
    @Published var passwordRepeat: String = ""

    var validatedEMail: AnyPublisher<String?, Never> {
        return $eMail
            .debounce(for: 0.5, scheduler: RunLoop.main)
            .removeDuplicates()
            .map { username in
                return Future { promise in
                    print("username: \(username)")
                    self.usernameAvailable(username) { available in
                        promise(.success(available ? username : nil))
                    }
                }
        }
        .switchToLatest()
            .eraseToAnyPublisher()
    }

    var validatedPassword: AnyPublisher<String?, Never> {
        return Publishers.CombineLatest($password, $passwordRepeat)
            .debounce(for: 0.5, scheduler: RunLoop.main)
            .map { password, passwordRepeat in
                print("password: \(password), passwordRepeat: \(passwordRepeat)")
                guard password == passwordRepeat, password.count > 5 else { return nil }
                return password
        }
        .eraseToAnyPublisher()
    }

    var validatedCredentials: AnyPublisher<(String, String)?, Never> {
        return Publishers.CombineLatest(validatedEMail, validatedPassword)
            .receive(on: RunLoop.main)
            .map { validatedEMail, validatedPassword in
                print("validatedEMail: \(validatedEMail ?? "not set"), validatedPassword: \(validatedPassword ?? "not set")")
                guard let eMail = validatedEMail, let password = validatedPassword else { return nil }
                return (eMail, password)
        }
        .eraseToAnyPublisher()
    }

    func usernameAvailable(_ username: String, completion: (Bool) -> Void) {
        let isValidEMailAddress: Bool = NSPredicate(format:"SELF MATCHES %@", "[A-Z0-9a-z._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,64}").evaluate(with: username)

        completion(isValidEMailAddress)
    }
}

#if DEBUG
struct RegistrationView_Previews : PreviewProvider {
    static var previews: some View {
        RegistrationView()
    }
}
#endif

Swift Combine's CombineLatest does not fire in response to an update to one of its publishers

You need to use Publishers.Merge rather than CombineLatest, see the documentation:

For Publishers.CombineLatest:

A publisher that receives and combines the latest elements from two publishers.

For Publishers.Merge

A publisher that emits an event when either upstream publisher emits an event.

Combine combineLatest is not waiting for the previous operation to fire

CombineLatest would have waited, as you correctly expected, but in your case getUsers already has a value ready, which was []; i.e what users was when getUsers ran.

You don't actually need to use CombineLatest to "wait" until some async action has happened. You can just chain publishers:

return importUsers()
           .setFailureType(to: Error.self)
           .flatMap { _ in
               getUsers()
           }
           .eraseToAnyPublisher()

In fact, getUsers isn't even needed, if you can assume that users is populated after importUsers:

return importUsers()
           .map { _ in
               self.users
           }
           .eraseToAnyPublisher()

CombineLatest operator is not emitting when inners publishers use subscribe(on:)

Yes, it's a bug. We can simplify the test case to this:

import Combine
import Dispatch

let pub = Just("x")
    .subscribe(on: DispatchQueue.main)

let ticket = pub.combineLatest(pub)
    .sink(
        receiveCompletion: { print($0) },
        receiveValue: { print($0) })

This never prints anything. But if you comment out the subscribe(on:) operator, it prints what's expected. If you leave subscribe(on:) in, but insert some print() operators, you'll see that the CombineLatest operator never sends any demand upstream.

I suggest you copy the CombineX reimplementation of CombineLatest and the utilities it needs to compile (the CombineX implementations of Lock and LockedAtomic, I think). I don't know that the CombineX version works either, but if it's buggy, at least you have the source and can try to fix it.

Using Combine to set up Publishers and while also satisfying requirement for initial values on non-nil variables

I created a new initializer struct for each that just includes the logic for each. Whenever I set the initial values, or update the published variables, I use this to avoid duplicating the logic. I'm open to other possibilities if someone has a better solution to this.

import Foundation
import Combine
import WatchConnectivity

class WatchConnectivityModel: ObservableObject {

    // values used to show/hide UI
    @Published var appNotInstalled: Bool
    @Published var complicationNotInstalled: Bool

    private struct AppNotInstalled {

        let value: Bool

        init(_ activationState: WCSessionActivationState,
             _ appInstalled: Bool) {
            value = !(activationState == .activated && appInstalled)
        }
    }

    private struct ComplicationNotInstalled {

        let value: Bool

        init(_ activationState: WCSessionActivationState,
             _ appInstalled: Bool,
             _ complicationInstalled: Bool) {
            value = activationState == .activated && appInstalled && !complicationInstalled
        }
    }

    private var cancellables: [AnyCancellable] = []

    init() {

        // initilize based on the current values
        let state = WCSession.default.activationState
        let appInstalled = WCSession.default.isWatchAppInstalled
        let complicationInstalled = WCSession.default.isComplicationEnabled

        appNotInstalled = AppNotInstalled(state,
                                          appInstalled).value
        complicationNotInstalled = ComplicationNotInstalled(state,
                                                            appInstalled,
                                                            complicationInstalled).value

        // set up the publishers
        let activationStatePublisher = WCSession.default
            .publisher(for: \.activationState)
        let isWatchAppInstalledPublisher = WCSession.default
            .publisher(for: \.isWatchAppInstalled)
        let isComplicationEnabledPublisher = WCSession.default
            .publisher(for: \.isComplicationEnabled)

        // set up assignment of appNotInstalled
        Publishers.CombineLatest(activationStatePublisher.removeDuplicates(),
                                 isWatchAppInstalledPublisher.removeDuplicates())
            .map { (state, installed) in
                return AppNotInstalled(state, installed).value
        }.receive(on: RunLoop.main)
            .assign(to: \.appNotInstalled,
                    on: self)
            .store(in: &cancellables)

        // set up assignment of complicationNotInstalled
        Publishers.CombineLatest3(activationStatePublisher.removeDuplicates(),
                                  isWatchAppInstalledPublisher.removeDuplicates(),
                                  isComplicationEnabledPublisher.removeDuplicates())
            .map { (state, appInstalled, complicationInstalled) in
                return ComplicationNotInstalled(state, appInstalled, complicationInstalled).value
        }.receive(on: RunLoop.main)
            .assign(to: \.complicationNotInstalled,
                    on: self)
            .store(in: &cancellables)
    }
}

How can I create a Swift Combine publisher from two publishers A and B where publisher B consumes the value from publisher A?

Assuming you want to start a new database publisher each time the Defaults publisher emits a change, you need the switchToLatest() operator.

This operator needs errors from both publishers to be harmonized. Here, since Defaults.publisher has the Never failure type, we can use the setFailureType(to:) operator in order to converge on the database publisher failure type: Error.

This gives:

func tasksPublisher() -> AnyPublisher<[Task], Error> {
    Defaults
        .publisher(.myUserDefault)
        .setFailureType(to: Error.self)
        .map({ change -> DatabasePublishers.Value<[Task]> in
            let myUserDefault = change.newValue
            return ValueObservation
                .tracking { db in
                    try Task
                        .someFilter(myUserDefault)
                        .fetchAll(db)
                }
                .removeDuplicates()
                .publisher(in: database)
        })
        .switchToLatest()
        .eraseToAnyPublisher()
}

Note that the returned publisher has the Error failure type, because the database is not 100% reliable, as all I/O externalities. It is difficult, in a Stack Overflow answer, to recommend hiding errors at this point (by turning them into an empty Task array, for example), because hiding errors prevents your app from knowing what's wrong and react accordingly.

Yet here is a version below that traps on database errors. This is the version I would use, assuming the app just can't run when SQLite does not work: it's sometimes useless to pretend such low-level errors can be caught and processed in a user-friendly way.

// Traps on database error
func tasksPublisher() -> AnyPublisher<[Task], Never> {
    Defaults
        .publisher(.myUserDefault)
        .map({ change -> AnyPublisher<[Task], Never> in
            let myUserDefault = change.newValue
            return ValueObservation
                .tracking { db in
                    try Task
                        .someFilter(myUserDefault)
                        .fetchAll(db)
                }
                .removeDuplicates()
                .publisher(in: database)
                .assertNoFailure("Unexpected database failure")
                .eraseToAnyPublisher()
        })
        .switchToLatest()
        .eraseToAnyPublisher()
}

Swift Combine: Subsequent Publisher That Consumes Other Publishers (Using Combinelatest) Doesn't "Fire"