Rx (or Combine) wrapper for Sharing library

[marcuswu0814]

Hi all,

In my current project, many of our services expose state through Observables as their interface. As a result, I often find myself writing a lot of boilerplate code to bridge these observables into TCA bindings. Here’s a simplified example:

@Reducer
struct MyFeature {
  @ObservableState
  struct State: Equatable {
    var myStatus: Status?
  }

  enum Action: BindableAction {
    case binding(BindingAction<State>)
    case task
  }

  enum CancelID { case task }

  @Dependency(\.someService) var service

  var body: some Reducer<State, Action> {
    Reduce { state, action in
      switch action {
      case .task:
        return service
          .observableStatus()
          .toEffectAndSkipError() // Helper to convert Rx Observable to Effect
          .map { Action.binding(.set(\.myStatus, $0)) }
          .cancellable(id: CancelID.task, cancelInFlight: true)
      }
    }
  }
}

As the number of observables I need to listen to increases, this pattern leads to a lot of repetitive code.

To reduce this boilerplate, I tried adopting the Sharing library and created a custom key for Rx observables like this:

@preconcurrency import RxSwift
import Sharing

extension SharedReaderKey {
    static func rxObservable<Value>(
        _ observable: @autoclosure () -> Observable<Value?>
    ) -> Self where Self == RxObseravbleShareKey<Value>.Default {
        Self[RxObseravbleShareKey(observable: observable()), default: nil]
    }
}

struct RxObseravbleShareKey<Element>: SharedReaderKey {
    typealias Value = Element?
    var id: some Hashable { UUID() }

    let observable: Observable<Value>

    func load(context: LoadContext<Value>, continuation: LoadContinuation<Value>) {
        Task {
            await continuation.resume(
                with: Result { try await observable.take(1).asSingle().value }
            )
        }
    }

    func subscribe(
        context: LoadContext<Value>,
        subscriber: SharedSubscriber<Value>
    ) -> SharedSubscription {
        let disposable = observable
            .subscribe(onNext: {
                subscriber.yield($0)
            })

        return .init {
            disposable.dispose()
        }
    }
}

Now I can use it like this:

struct State: Equatable {
  @SharedReader(.rxObservable({
    @Dependency(\.someService) var service
    return service.observableStatus()
  }))
  var myStatus: Status?
}

This works well so far, but I’d love to get feedback:
• Are there any trade-offs or pitfalls I should be aware of when using this approach?
• Is there a better or more idiomatic way to bridge RxSwift.Observable into Sharing?

Any insights or suggestions would be greatly appreciated. Thanks! 🙏

[kamcma]

One thing that gives me pause is the random ID for the key.

Take for instance the AppStorageKey shipped with the library. Its ID is a combination of its dependencies (a particular instance of UserDefaults) and its parameters (the string key). That way, if you ask for the same shared value in multiple places, meaning the same dependencies and the same parameters, the library can hand you the same shared value and not spin up multiple instances of observation of the same thing.

Your key, on the other hand, I think would redundantly observe the same thing because each instance of the key is getting its own, random ID. Fixing this might not be possible with this key, because its parameter is an unhashable closure that can be and do
literally anything. I think that's why the keys shipping with the library are not quite that generic, so that they can hash their dependencies and parameters.

Depending on how you're using it, you may be able to tolerate the inefficiency. But that's what catches my eye.