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! 🙏