https://github.com/cats-oss/Unio で公開しているFrameworkの動きを簡易的に表現しています

UnioSystem.swift

// Playgroundでの実行を想定しているのでCombineを利用
import Combine

// MARK: - CombineをRxSwiftのように定義する
typealias Observable<T> = AnyPublisher<T, Error>
typealias PublishRelay<T> = PassthroughSubject<T, Never>
typealias BehaviorRelay<T> = CurrentValueSubject<T, Never>
typealias DisposeBag = [AnyCancellable]

// MARK: - Output関連

/// Outputを定義していることを明示するためのprotocol
protocol OutputType {}

/// subjectでイベントを発生させるメソッドなどにはアクセス不可とし、Observableやvalueにのみアクセスできるようにした型
@dynamicMemberLookup
final class OutputWrapper<Output: OutputType> {
    private let output: Output

    init(_ output: Output) {
        self.output = output
    }

    /// dynamicMemberLookupでBehaviorRelayのObservableにのみアクセスできるようにしている
    subscript<U, S: BehaviorRelay<U>>(dynamicMember keyPath: KeyPath<Output, S>) -> Observable<S.Output> {
        output[keyPath: keyPath]
            .catch { _ -> Observable<S.Output> in fatalError() } // 型合わせのために記載しているので実行されない
            .eraseToAnyPublisher()
    }

    /// dynamicMemberLookupでBehaviorRelayのvalueにのみアクセスできるようにしている
    subscript<U, S: BehaviorRelay<U>>(dynamicMember keyPath: KeyPath<Output, S>) -> S.Output {
        output[keyPath: keyPath].value
    }
}

// MARK: - Input関連

/// Inputを定義していることを明示するためのprotocol
protocol InputType {}

/// subjectでObservableやvalueなどにはアクセス不可とし、イベントを発生させるメソッドのみアクセスできるようにした型
@dynamicMemberLookup
final class InputWrapper<Input: InputType> {
    private let input: Input

    init(_ input: Input) {
        self.input = input
    }

    /// dynamicMemberLookupでPublishRelayのイベントを発生させるメソッドにアクセスできるようにしている
    subscript<U, S: PublishRelay<U>>(dynamicMember keyPath: KeyPath<Input, S>) -> (S.Output) -> Void {
        { [input] in
            input[keyPath: keyPath].send($0)
        }
    }
}

// MARK: - Logic周り

/// Inputをバインドして、Outputを生成することを明示する型
protocol LogicType {
    associatedtype Input: InputType
    associatedtype Output: OutputType
    static func bind(input: Input, disposeBag: inout DisposeBag) -> Output
}

// MARK: - UnioStream周り

/// inputとoutputの初期化を担った型
class PrimitiveStream<Logic: LogicType> {
    typealias Input = Logic.Input
    typealias Output = Logic.Output

    let input: InputWrapper<Input>
    let output: OutputWrapper<Output>
    private var disposeBag: DisposeBag = []

    deinit { print("deinit") }

    init(input: Input, logic _: Logic.Type) {
        self.input = InputWrapper(input)
        self.output = OutputWrapper(Logic.bind(input: input, disposeBag: &disposeBag))
    }
}

/// PrimitiveStreamにLogicTypeを実装することを強制するための型
/// - note: PrimitiveStreamをinputとoutputの初期化を担った抽象クラスのように扱い
///         bindはサブクラス側に実装を強制したいため、PrimitiveStreamには直接採用しない
typealias UnioStream<Logic: LogicType> = PrimitiveStream<Logic> & LogicType

// MARK: - アプリ側の実装例

final class CountStream: UnioStream<CountStream> {
    struct Input: InputType {
        let countUp = PublishRelay<Void>()
        let countDown = PublishRelay<Void>()
    }

    struct Output: OutputType {
        let count: BehaviorRelay<Int>
    }

    init() {
        super.init(input: Input(), logic: CountStream.self)
    }

    static func bind(input: Input, disposeBag: inout DisposeBag) -> Output {
        // 簡易化するためにStateを設けていないので、countをStateと位置づけます
        let count = BehaviorRelay<Int>(0)

        input.countDown.map { -1 }
            .merge(with: input.countUp.map { 1 })
            .flatMap { newValue -> Just<(Int, Int)> in // 実質withLatestFrom
                Just<(Int, Int)>((newValue, count.value))
            }
            .map { $0 + $1 }
            .assign(to: \.value, on: count)
            .store(in: &disposeBag)

        return Output(count: count)
    }
}

func main() {
    let countStream = CountStream()

    // OutputWrapper経由なので、型推論したObservableにアクセスしている
    let disposable = countStream.output.count
        .sink(receiveCompletion: { _ in }) { value in
            print("subscribe: \(value)")
        }
    defer { disposable.cancel() }

    // OutputWrapper経由なので、型推論したIntにアクセスしている
    print("value: \(countStream.output.count as Int)") // 0

    // InputWWrapper経由なので、型推論した(Void) -> Voidにアクセスしている
    countStream.input.countUp(())
    // OutputWrapper経由なので、型推論したIntにアクセスしている
    print("value: \(countStream.output.count as Int)") // 1

    // InputWWrapper経由なので、型推論した(Void) -> Voidにアクセスしている
    countStream.input.countDown(())
    // OutputWrapper経由なので、型推論したIntにアクセスしている
    print("value: \(countStream.output.count as Int)") // 0
}

main()