Using '.run' and 'await in' to observe AnyPublisher<Value, Never> does not work

[SebastianBoldt]

Hey,

I currently do have some Issues related to long living effects.

To give you some background:
I created a Dependency to observe a specific value from the UserDefaults.
It returns an AnyPublisher we can subscribe to, to get notified when the value changes.

enum CardMode: String, Codable, CaseIterable {
    case expanded
    case collapsed
}

struct CardModeProvider {
    var cardMode: () -> CardMode
    var cardModePublisher: AnyPublisher<CardMode, Never>
    var storeCardMode: (CardMode) -> Void
}

extension CardModeProvider {
    static let live: CardModeProvider = {
        return Self(
            cardMode: {
                CardMode(rawValue: UserDefaults.sharedGroupDefaults.cardMode) ?? .expanded
            },
            cardModePublisher: UserDefaults.sharedGroupDefaults
                .publisher(for: \.cardMode)
                .map { CardMode(rawValue: $0) ?? .expanded }
                .eraseToAnyPublisher(),
            storeCardMode: {
                UserDefaults.sharedGroupDefaults.cardMode = $0.rawValue
            }
        )
    }()
}

private enum CardModeDependencyKey: DependencyKey {
    static let liveValue = CardModeProvider.live
}

extension DependencyValues {
  var cardModeProvider: CardModeProvider {
    get { self[CardModeDependencyKey.self] }
    set { self[CardModeDependencyKey.self] = newValue }
  }
}

Inside the reducer I observed the value changes using eraseToEffect:

func reduce(into state: inout State, action: Action) -> EffectTask<Action> {
    switch action {
        case .onAppear:
            return cardModeProvider.cardModePublisher.eraseToEffect {
                .updateCardMode($0)
            }
    }
}

it works like a charm but after I took a look into the Docs I discovered that it is recommended to use .run and for await value in sequence for that. So I tried to do that.

.run { send in
    for await value in cardModeProvider.cardModePublisher.values {
        await send(.updateCardMode(value))
    }
}

But unfortunately I doesn't work as expected.
It will get triggered once, but each time I change the value in the User-Defaults the action will not get send.
I first assumed that my publisher was cancelled or completed somehow under the hood so I tried to investigate using handleEvents.

static let live: CardModeProvider = {
        return Self(
            cardMode: {
                CardMode(rawValue: UserDefaults.sharedGroupDefaults.cardMode) ?? .expanded
            },
            cardModePublisher: UserDefaults.sharedGroupDefaults
                .publisher(for: \.cardMode)
                .handleEvents(receiveCompletion: {
                    // ...
                }, receiveCancel: {
                    // ...
                })
                .map { CardMode(rawValue: $0) ?? .expanded }
                .eraseToAnyPublisher(),
            storeCardMode: {
                UserDefaults.sharedGroupDefaults.cardMode = $0.rawValue
            }
        )
  }()

But it turned out that none of those publisher events occur.
Does anyone maybe has an idea what I am doing wrong?

[tgrapperon]

Hey @SebastianBoldt! If UserDefaults.sharedGroupDefaults is a stable instance (a singleton for example), the async loop you presented should produce new values/actions if the publisher is sound.
BTW, you have the .print() method that works on Publisher, an that shows various informations about the subscription. You can maybe try to loop over cardModeProvider.cardModePublisher.print().values to see if it emits or not.
I would also write a test for the cardModeProvider publisher, to confirm that it works as expected in isolation. This publisher is KVO-based, but it should work for UserDefaults (that's what I'm using in swift-dependencies-additions.
If nothing stands out, I would try to perform read/write on the MainActor for example, as this is something that could be different between isolated testing and TCA.

[SebastianBoldt]

Do you have any Idea why? Is it an Issue on my side?

[SebastianBoldt]

There has to be a difference between using eraseToEffect to observe the changes and the for await value in sequence way 🤔

[tgrapperon]

I don't think that this is an issue from your side! This is an odd behavior of the KVO Publisher (and maybe other ones) with respect to back pressure I guess. Using values on these publishers doesn't seem to produce the same values as sinking on the publisher. This is maybe a bug, or maybe the expected behavior. I'm considering submitting a PR about this.
A unit test over the AsyncSequence would have spotted the issue, but these tests can be a little tricky to set up (as you saw from the combine version). Maybe some test helper for async sequences could help.

[iampatbrown]

Maybe @iampatbrown has an idea?

@SebastianBoldt @tgrapperon I'm pretty sure this is a KeyValueObservingPublisher bug similar to the one with ObservableObjectPublisher. The workaround is to use Publishers.Buffer eg:

 cardModePublisher: UserDefaults.sharedGroupDefaults
  .publisher(for: \.cardMode,)
  .buffer(size: 1, prefetch: .byRequest, whenFull: .dropOldest)
  .map { CardMode(rawValue: $0) ?? .expanded }
  .eraseToAnyPublisher(),

[tgrapperon]

Thanks @iampatbrown! This may be a better solution than my handleEvents solution because I found that mine is changing the initial value behavior and doesn't emit it anymore when we specify .request(.unlimited).