What is the best practice to set up a long-live publisher into a reducer without duplication?

[ylorn]

The Prerequisite

I am working with a 3rd party API that provides a Publisher that constantly emits values through the app's life cycle. To mimic its behavior and simplify the demo, I have come up with the following code piece:

import Combine

class APIClient {
    static let shared: APIClient = .init()

    let subject: PassthroughSubject<Int, Never> = .init()
    private var cancellables: Set<AnyCancellable> = []

    private init() {
        Timer.publish(every: 1, on: .main, in: .default)
            .autoconnect()
            .map { _ in Int.random(in: 0 ..< 999 )}
            .sink(receiveValue: self.subject.send)
            .store(in: &self.cancellables)
    }
}

Basically, the API I am working with is the APIClient.subject, which is a long-live publisher that keeps emitting Int values. To start with TCA, I first implement the DependecyValues to make the APIClient globally available:

import Dependencies

extension DependencyValues {
    var apiClient: APIClient {
        get { self[APIClientKey.self] }
        set { self[APIClientKey.self] = newValue }
    }

    private enum APIClientKey: DependencyKey {
        static var liveValue: APIClient = .shared
    }
}

Then I implement the ReducerProtocol:

import ComposableArchitecture

struct Demo: ReducerProtocol {
    struct State: Equatable {
        var output: Int
    }

    enum Action: Equatable {
        case viewAppear
        case update(output: Int)
    }

    @Dependency(\.apiClient) var apiClient

    var body: some ReducerProtocol<State, Action> {
        Reduce { state, action in
            switch action {
            case .viewAppear:
                return self.apiClient.subject
                    .map(Action.update(output:))
                    .eraseToEffect()

            case .update(let output):
                state.output = output
                return .none
            }
        }
    }
}

And finally, to composite the View:

import SwiftUI

struct DemoView: View {
    var store: StoreOf<Demo>

    var body: some View {
        WithViewStore(self.store) { viewStore in
            Text("\(viewStore.state.output)")
                .onAppear {
                    viewStore.send(.viewAppear)
                }
        }
    }
}

struct DemoView_Previews: PreviewProvider {
    static var previews: some View {
        DemoView(
            store: .init(
                initialState: Demo.State(output: 0),
                reducer: Demo()
            )
        )
    }
}

The effect of the DemoView is simple, every time the Timer ticks, a random Int value is updated on the center of the view.

Questions

1. The EffectTask to combine the APIClient.subject to Demo.State.output is created every time the .viewAppear action is received. Will multiple identical EffectTask be created in the TCA if .viewAppear action is sent numerous times through out the app's life cycle? If so, what is the best practice to avoid it?
2. Currently, the bind between APIClient.subject and Demo.State.output is only created when the DemoView appears. But what I want is to permanently bind these 2 parties without having to manually do so in Demo.body and explicitly trigger the binding in DemoView.onAppear with a dedicated Action. Because the Demo.State.output should always reflect the latest value from APIClient.subject, no matter the DemoView appears or not. What is the best way to achieve so? Is there an API to accomplish this like:

struct Demo: ReducerProtocol {
   struct State: Equatable {
       var output: Int
   }

   enum Action: Equatable {
-        case viewAppear
-        case update(output: Int)
   }

   @Dependency(\.apiClient) var apiClient

   var body: some ReducerProtocol<State, Action> {
       Reduce { state, action in
-            switch action {
-            case .viewAppear:
-                return self.apiClient.subject
-                    .map(Action.update(output:))
-                    .eraseToEffect()
-
-            case .update(let output):
-                state.output = output
-                return .none
-            }
       }
+       .onceAndForAll { state in
+           return self.apiClient.subject
+               .assign(to: \.output, on: state)
+               .eraseToEffect()
+        }
   }
}

struct DemoView: View {
   var store: StoreOf<Demo>

   var body: some View {
       WithViewStore(self.store) { viewStore in
           Text("\(viewStore.state.output)")
-                .onAppear {
-                    viewStore.send(.viewAppear)
-                }
       }
   }
}

In which the .onceAndForAll {...} function sets the binding permanently for only once no matter how many times it gets called.

3. If I want to completely get rid of the dependency of Combine in this case, what is the best practice to wrap the original API to the structured asynchrony and how to bridge that into the new ReducerProtocol? Via AsyncStream?

Any help is appreciated, thanks in advance!

[tgrapperon]

Hey @ylorn, thanks for the very nice case presentation. There is one thing that are not quite right in your implementation. In .viewAppear, you create a subscription to the apiClient, but you're not holding any reference to it. If this view disappears, the subscription will still be active and will start to emit actions for this non-existing Demo domain. This is considered as a programming error in TCA. Furthermore, if it appears again, it will indeed create a new subscription. There are two ways to fix this issue:

• You need an identifier to tag this subscription. The most convenient way is to introduce an empty enum and use this type as an identifier itself. You can sometimes use the reducer type itself for this purpose in simple domains with one subscription. You then use this value to tag your subscription as .cancellable() and you explicitly cancel it in onDisappear:

enum ApiObservationID {}
var body: some ReducerProtocol<State, Action> {
  Reduce { state, action in
    switch action {
    case .viewAppear:
      return self.apiClient.subject
        .map(Action.update(output:))
        .eraseToEffect()
        .cancellable(id: ApiObservationID.self)

    case .viewDisappear:
      return .cancel(id: ApiObservationID.self)
    …

This can be problematic however if Demo.State? is conditioning the appearance of DemoView, as you will likely receive an action form the subscription while your state is nil but onDisappear not yet called. The workaround is to handle the subscription cancellation from the parent, but this is very cumbersome.

• The other and preferred approach is to harness Swift concurrency and use a .task modifier in your view that will condition the lifetime of the effect and will cancel it automatically when it will cancel the .task when the view will disappear. You can then remove viewAppear and introduce a task action:

struct Demo: ReducerProtocol {
  struct State: Equatable {
    var output: Int
  }

  enum Action: Equatable {
    case task
    case update(output: Int)
  }

  @Dependency(\.apiClient) var apiClient
  var body: some ReducerProtocol<State, Action> {
    Reduce { state, action in
      switch action {
      case .task:
        return .run { send in
          for await output in self.apiClient.subject.values {
            await send(.update(output: output))
          }
        }
      case .update(let output):
        state.output = output
        return .none
      }
    }
  }
}

struct DemoView: View {
  var store: StoreOf<Demo>

  var body: some View {
    WithViewStore(self.store) { viewStore in
      Text("\(viewStore.state.output)")
        .task {
          await viewStore.send(.task).finish()
        }
    }
  }
}

Please note the .finish() in View.task that makes the send an async function that returns when the corresponding EffectTask finishes (here never). So when this task is cancelled, it will in turn cancel the .run EffectTask. Please also note that I've used the .value property on Publisher that makes it an AsyncSequence of its Output (it requires iOS 15+). This async/await approch suffers the same problems as the Combine one regarding presentation, when your DemoView visibility is conditioned by Demo.State? being nil or not. This is will be addressed in the upcoming navigation update.

It is probably fine to have a dozen of views subscribing to your dependency at the same time. Please let us know if you have a specific case where you have much more features or real-life performance issues, as the solution is then probably bespoke. I don't know how it exactly works, but the onceAndForAll you're presenting to create a new subscription for each action received, which will be very problematic.

Because the Demo.State.output should always reflect the latest value from APIClient.subject, no matter the DemoView appears or not.

This is not really the way TCA is expected to work: If Demo isn't alive (that is its state is nil), the DemoView shouldn't be presented. If one can imagine having a non-nil Demo.State in your model without a corresponding DemoView, this is probably a sign that you could design your model differently and have this feature controlled at a higher level, for example in a View that doesn't disappear. Please also note that because your subject is a PassThroughSubject, it won't keep the current value, and you can probably convert it to a CurrentValueSubject if you can provide an initial value.

If I want to completely get rid of the dependency of Combine in this case, what is the best practice to wrap the original API to the structured asynchrony

Using an ad-hoc AsyncStream is a solution, or Publisher.values if you can drop iOS 14. There are also backports on GitHub.
If your only requirement is to have a Timer, you can even use the shiny-new swift-clocks, but I understand that your APIClient is a mock.

Please let me know if this helps with your problem.

[ylorn]

@tgrapperon Thanks a lot for the detailed explanation and solution walkthrough! It gives me a better understanding of TCA!

I have several questions about your response though:

1. About the tagging and cancellation in .viewAppear and .viewDisappear

You need an identifier to tag this subscription. The most convenient way is to introduce an empty enum and use this type as an identifier itself. You can sometimes use the reducer type itself for this purpose in simple domains with one subscription. You then use this value to tag your subscription as .cancellable() and you explicitly cancel it in onDisappear:

is it possible to have an option to override/skip the subscription if its identifier already exists? like

enum ApiObservationID {}
var body: some ReducerProtocol<State, Action> {
  Reduce { state, action in
    switch action {
    case .viewAppear:
      return self.apiClient.subject
        .map(Action.update(output:))
        .eraseToEffect()
        .cancellable(
            id: ApiObservationID.self,
+           option: .skipOnExisting // or .overrideOnExisting
        )

-    case .viewDisappear:
-      return .cancel(id: ApiObservationID.self)

So we can avoid having multiple subscription for the same instance. Is this a viable way?

2. About the .task alternative:
   This looks like a very common pattern to have in any view. Is it possible to have a dedicated ReducerBuilder and a ViewStore to eliminate the boilerplate? Like:

    enum Action: Equatable {
-        case task
-        case update(output: Int)
    }

    @Dependency(\.apiClient) var apiClient

    var body: some ReducerProtocol<State, Action> {
+
+        Task { state in
+            for await value in self.apiClient.api1.values {
+                state.output1 = value
+            }
+        }
+
+        Task { state in
+            for await value in self.apiClient.api2.values {
+                state.output2 = value
+            }
+        }
+
+        Task { state in
+            for await value in self.apiClient.api3.values {
+                state.output3 = value
+            }
+        }
+
+        ...
    }

...

struct DemoView: View {
    var store: StoreOf<Demo>

    var body: some View {
-        WithViewStore(self.store) { viewStore in
-            Text("\(viewStore.state.output)")
-                .task {
-                    await viewStore.send(.task).finish()
-                }
-        }

+        WithTaskViewStore(self.store) { taskViewStore in
+            Text("\(taskViewStore.state.output1)")
+            Text("\(taskViewStore.state.output2)")
+            Text("\(taskViewStore.state.output3)")
+            ...
+        }
    }
}

3. About the .onceAndForAll function.

but the onceAndForAll you're presenting to create a new subscription for each action received, which will be very problematic.

Sorry, I was thinking about the Reducer type before the ReducerProtocol when I was proposing this. Because then we are using the Reducer as a global variable like let reducer = Reducer<State, Action, Environment> { state, action, environment in ... }, which is shared by all Store instances that requires it. Therefore I am thinking we can have the onceAndForAll modifier function to apply the observation only once, and that observation will be also shared to update every State instance when the mutable inout state is passed into the reducer.

I am going to give it a go with your suggestions and see where it leads me.
Many many thanks for helping out!

[tgrapperon]

is it possible to have an option to override/skip the subscription if its identifier already exists? like

There is currently no way to poll existing subscriptions (or more precisely, cancellable effects) from the reducer and perform some logic. There is a cancelInFlight: Bool flag in cancellable(…) that will make this effect cancel and replace any existing effect with the same ID, but there is nothing to perform the opposite (nor really a need for it). Subscriptions are usually very cheap if that bothers you, and you can cancel them as soon as you don't need them, and recreate them when you need. If this is problematic, feel free to get back with a more explicit use-case, so we can provide better guidance. At worse, you can subscribe from a parent that doesn't appear/disappear, and send the actions from there, like a notification in some sense.

This looks like a very common pattern to have in any view. Is it possible to have a dedicated ReducerBuilder and a ViewStore to eliminate the boilerplate?

This is very common, but the boilerplate will likely always be smaller than a generalized formalization that will probably be either general enough but too complex to use, or the opposite. During the "Concurrency" beta, I suggested adding a method to WithViewStore, so we would be able to only specify the action that we want to send:

WithViewStore(store, observe: {…}) { 
  … 
}.task(.someAction)

It would only allow you to save a few characters, but it could help improving the discovery of this pattern, so maybe it's worth it. I don't know. It's very easy to add too much sugar to a library. Maybe I'll reconsider when the navigation beta will begin, as this pattern will most likely play nicely with it.

Sorry, I was thinking about the Reducer type before the ReducerProtocol

Sorry, I missed that indeed. With ReducerProtocol, reducers are currently stateless and you can't store information that will persist between actions being received. It can be improved in the future, and one could even imagine something similar to SwiftUI's onReceive at some point. I think that for now, the simplest approach to your problem is to subscribe to the publisher where you need it. Again, if this is problematic in your real-life situation, feel free to get back with more details about it.

[tgrapperon]

BTW, if you go the AsyncStream route, make sure to return a new AsyncStream value each you pull one out from a @Dependency (in other words, model it as a computed property of APIClient like you did, or as a standalone function () -> AsyncStream<Output>, but not as standalone AsyncStream<Output> property directly out of DependencyValues, as these top-level values are now cached and the same value is returned each time). This is because iterating over the same AsyncSequence from different places produces unexpected results, whereas it is fine to do so when you subscribe to a PassthroughSubject.

[ylorn]

@tgrapperon Would you mind giving a code piece with the APIClient.subject for an example to use AsyncStream?

[tgrapperon]

For example:

class APIClient {
  static let shared: APIClient = .init()

  // We use `AsyncStream` as a pseudo `AnyAsyncSequence`
  // that doesn't exist.
  // The initializer is provided by the TCA and uses here
  // the `.values` property from iOS 15.
  var ticks: AsyncStream<Int> {
    AsyncStream(subject.values)
  }

  var ticks2: AsyncStream<Int> {
	// If you can't use iOS 15, you can use this backport
    // from [John Sundell](https://github.com/JohnSundell/AsyncCompatibilityKit)
    AsyncStream { continuation in
      var cancellable: AnyCancellable?
      let onTermination = { cancellable?.cancel() }
      
      continuation.onTermination = { @Sendable _ in
        onTermination()
      }

      cancellable = subject.sink(
        receiveCompletion: { _ in
          continuation.finish()
        },
        receiveValue: { value in
          continuation.yield(value)
        }
      )
    }
  }

  let subject: PassthroughSubject<Int, Never> = .init()
  private var cancellables: Set<AnyCancellable> = []

  private init() {
    Timer.publish(every: 1, on: .main, in: .default)
      .autoconnect()
      .map { _ in Int.random(in: 0..<999) }
      .sink(receiveValue: self.subject.send)
      .store(in: &self.cancellables)
  }
}

and in the reducer:

return .run { send in
  for await tick in apiClient.ticks {
    await send(.update(output: tick))
  }
}

Hope it'll help!

[ylorn]

The ticks2's implementation saved my life, I can't just drop the iOS 14 support yet. You rock!