Handling long-running observers

[damirstuhec]

I'm wondering how should we approach representing long-running observers as effects?

For example, NotificationCenter.default.publisher never completes which, I believe, makes it hard to represent as an Effect of composable architecture.

As a workaround, I'm currently creating and observing NotificationCenter's publisher in the SceneDelegate, and then sending an appropriate action into the store directly, since I have access to it from there.

That works, but it doesn't feel right and makes testing very hard. Any input would be greatly appreciated.

[mbrandonw]

This is a very good question, and something important to attack.

It takes some time to develop the machinery necessary to attack this problem, but I can sketch a bit of what is necessary.

First, ideally everything is done in the reducer instead of the scene delete. This makes it testable and more understandable. But, as you said, it's difficult to do with long living effects.

The idea is that you have a "kick-off" action that starts up the effect, such as onAppear in SwiftUI or viewWillAppear in UIKit. Then you have a tear down action to stop the effect, such as onDisappear or viewDidDisappear.

And the way you can stop an effect is to introduce an operator that enhances an effect with the capability of being canceled, as well as a new effect that is capable of canceling an in-flight cancellable effect. This gist does that:

https://gist.github.com/mbrandonw/1c4ca76f1927420921b3871dd2153497

(also a fellow Point-Free subscriber @alexito4 made a nice video on using this code http://youtube.com/watch?v=VAB3lysXU9o)

Once you have those capabilities you can do things like this in your reducer:

struct CancelNotificationToken: Hashable { static let shared = CancelNotificationToken() }

case .onAppear:
  // Start listening for notifications
  return [
    NotificationCenter.default.publisher
      .map(MyAction.receivedNotification)
      .eraseToEffect()
  ]

case .onDisappear:
  // Stop listening for notifications
  return [
    Effect.cancel(id: CancelNotificationToken.shared)
      .fireAndForget(),
  ]

That's the basics. The ergonomics could probably be improved a bit, and there may be some edge cases to think through (especially because we haven't done this with notification center yet, so there may be more to do for this case), but this is how you can start listening for long-living effects and then shut them down when you are done.

Hope it helps!

[mayoff]

My approach to this problem is based on what Elm calls a "subscription". I call it an "extrinsic" in my implementation to avoid confusion with Combine's Subscription. You can find my complete implementation in this gist.

Elm subscriptions work like this:

• Elm has an opaque type Sub that represents a persistent source of external events.

• Various Elm kernel packages provide functions to construct subscriptions. This includes things like Time.every and Browser.onKeyPress.

• When you create your Platform.Program using a function like Browser.element, you give it a subscriptions function. The subscriptions function takes your model as an argument and returns a Sub.

• Elm provides Sub.none and Sub.batch, so the one Sub your subscriptions function returns can actually be any number of subscriptions, including zero.

• After every model update, the Elm kernel calls your subscriptions function on the new model and intelligently starts/stops subscriptions as needed.

I implemented a similar API in my version of the Composable Architecture as follows:

• I defined a protocol Extrinsic:

  public protocol Extrinsic: Hashable {
    associatedtype Action
    func publisher() -> AnyPublisher<Action, Never>
  }

  It's a subprotocol of Hashable so that the Store can efficiently determine which extrinics are added and removed after each model update.

• I modified the root Store constructor to be parameterized by a concrete Extrinsic type, and to take an extrinsics function that returns the desired extrinsics for a model value. In my implementation, the constructor is a static method:

  public static func root<Ex: Extrinsic>(
      withModel model: Model,
      reducer: Reducer<Model, Action>,
      extrinsics: @escaping (Model) -> Set<Ex>
  ) -> Store<Model, Action> where Ex.Action == Action {
      return RootStore(
          model: model,
          reducer: reducer,
          extrinsicCollection: ExtrinsicCollection(extrinsics))
  }

• The ExtrinsicCollection class holds the extrinsics method and manages the Combine Subscriptions to the current set of extrinsics. It is generic over both the model type and the extrinsic type, but it subclasses ExtrinsicCollectionBase, which is only generic over the model and action types:

  fileprivate class ExtrinsicCollectionBase<Model, Action> {
      func update(with store: RootStore<Model, Action>) {
          fatalError("subclass responsibility")
      }
  }
  
  fileprivate class ExtrinsicCollection<Model, Ex: Extrinsic>: ExtrinsicCollectionBase<Model, Ex.Action> {
      init(_ make: @escaping (Model) -> Set<Ex>) {
          self.make = make
      }
  
      override func update(with store: RootStore<Model, Ex.Action>) {
          let exs = make(store.model)
          for ex in ticketForEx.keys.filter({ !exs.contains($0) }) {
              ticketForEx.removeValue(forKey: ex)?.cancel()
          }
          for ex in exs.subtracting(ticketForEx.keys) {
              ticketForEx[ex] = ex.publisher()
                  .sink(receiveValue: { [weak store] in store?.send($0) })
          }
      }
  
      private let make: (Model) -> Set<Ex>
      private var ticketForEx: [Ex: AnyCancellable] = [:]
  }

• The root Store has a property that holds its ExtrinsicCollection. Since ExtrinsicCollectionBase is generic over the same types as Store (and not over the Extrinsic type), the Store doesn't have to be generic over the type implementing Extrinsic:

  private let extrinsicCollection: ExtrinsicCollectionBase<Model, Action>?

• Every time the Store handles an action, after running effects, it updates its extrinsicCollection:

  override func send(_ action: Action) {
      let effects = reducer.apply(action, to: &_model)
      for effect in effects {
          var ticket: AnyCancellable? = nil
          ticket = effect.sink(
              receiveCompletion: { _ in ticket = nil },
              receiveValue: { self.send($0) })
          withExtendedLifetime(ticket) { }
      }
      extrinsicCollection?.update(with: self)
      subject.send()
  }

Here's what it looks like in use. Let's say I have this (ridiculous) model:

struct MyModel {
    var timers: [String: TimeInterval] = [:]
    var dings: [String] = []
}

The user can add and remove repeating timers with names. Every time a timer fires, I add its name to dings. But if the current AVAudioSession gets interrupted, I want to remove all timers and clear dings. Here are my actions and my reducer:

extension MyModel {
    enum Action {
        case setTimer(String, TimeInterval)
        case removeTimer(String)

        case timerFired(String)
        case audioSessionInterrupted
    }

    mutating func apply(_ action: Action) -> [Effect<Action>] {
        switch action {
        case .setTimer(let name, let interval):
            timers[name] = interval
        case .removeTimer(let name):
            timers.removeValue(forKey: name)
        case .timerFired(let name):
            dings.append(name)
        case .audioSessionInterrupted:
            timers = [:]
            dings = []
        }
        return []
    }
}

I define an extrinsic for a repeating timer and another for audio session interruptions:

extension MyModel {
    enum Extrinsics: Extrinsic {
        case timer(String, TimeInterval)
        case audioSessionInterruptions

        func publisher() -> AnyPublisher<Action, Never> {
            switch self {
            case .timer(let name, let interval):
                return Timer.publish(every: interval, on: .main, in: .common)
                    .map { _ in Action.timerFired(name) }
                    .eraseToAnyPublisher()

            case .audioSessionInterruptions:
                return NotificationCenter.default
                    .publisher(for: .AVAudioSessionInterruption)
                    .map { _ in Action.audioSessionInterrupted }
                    .eraseToAnyPublisher()
            }
        }
    }

    func extrinsics() -> Set<Extrinsics> {
        var exs = Set(timers.map { Extrinsics.timer($0, $1) })
        if !timers.isEmpty {
            exs.insert(.audioSessionInterruptions)
        }
        return exs
    }
}

And then I create my Store like this:

let rootStore = Store<MyModel, MyModel.Action>.root(
    withModel: .init(timers: [:]),
    reducer: Reducer { $0.apply($1) },
    extrinsics: { $0.extrinsics() })

[mathieutozer]

How might I handle testing long running observers using the Step test architecture?

I have the following test, and I believe my mock callbacks are being called correctly as the trips array is assigned correctly and the state equality test passes, but the receive step fails with "Timed out waiting for the effect to complete".

  func testLoadAllTrips() {
    assert(
      initialValue: TripListView.ViewState(trips: [], selection: Set()),
      reducer: tripFeatureReducer,
      environment: Apollo(MockApollo<AllTripsQuery>(store: ApolloStore(cache: InMemoryNormalizedCache()), watchQuery: {
        return GraphQLResult<AllTripsQuery.Data>.init(data: .init(tripsList: .init(items: [.init(id: "geneva2020", name: "Geneva 2020", summary: "Trip to Geneva to rennovate apartment")])), errors: nil, source: .server, dependentKeys: nil)
      })),
      steps:
      Step(.send, .watchTrips) {
        $0.trips = []
      },
      Step(.receive, .tripsDidRefresh(.success([geneva]))) {
        $0.trips = [geneva]
      }
//      ,
//      Step(.send, .stopWatchingTrips) { _ in
//      }
    )
  }

Adding the final .stopWatchingTrips action fails because it of course hasn't finished the receive step, which makes sense.

Do we need a 3rd step type to represent cancellable?

[shawnkoh]

has there been any conclusions on the best practice for handling subscriptions like these?