CombineでReactiveSwiftのthrottle(while:on:)が使いたい

# 前書き
皆さん、ReactiveSwiftはご存知だろうか？

正直なところ、RxSwiftと比べてシェアは低く、Combineが登場した今、新たに導入することはあまりないと思う。

実際この記事を書いている時点で、QiitaのRxSwiftの記事は562件あるのに対し、ReactiveSwiftの記事は13件しかなく、ReactiveSwiftのシェアの低さを物語っている。

しかし、ReactiveSwiftには他のReactiveProgrammingのライブラリにはない便利なオペレーターが存在する。

今回紹介する`throttle(while:on:)`もそのひとつ。

`while`に指定した`Property`の値が`true`の間、上流から来る値を堰き止め、`false`になったタイミングで最後に流れてきた値を下流に流す。

ReactiveSwiftのオペレーターについては@sei_kataokaさんの記事が非常に分かりやすいのでおすすめだ。

https://qiita.com/sei_kataoka/items/8e401e9d5b55ff40f3ed#throttlewhile-shouldthrottle-on-scheduler

今回はこれをCombineのオペレーターとして再現してみた。

# 方針
Combineに初めから実装されているオペレータは全て[Publishers](https://developer.apple.com/documentation/combine/publishers)に[Pubisher](https://developer.apple.com/documentation/combine/publisher)を準拠した`struct`や`class`の形で機能が実装されている。

例えば、よく使われるオペレーターの[Publisher.map(_:)](https://developer.apple.com/documentation/combine/publisher/map(_:)-99evh)は戻り値が[Publishers.Map](https://developer.apple.com/documentation/combine/publishers/map)になっている。

```Swift
extension Publisher {
    func map<T>(_ transform: @escaping (Self.Output) -> T) -> Publishers.Map<Self, T> {
        Publishers.Map(upstream: self, transform: transform)
    }
}
```

これは`map(_:)`の実際の処理が`Publishers.Map`に書かれているためで、実際のコードを見た訳ではないが`Publisher.map(_:)`では元の`Publisher`を`Publishers.Map`に置き換えているだけだと思われる。

今回はこれと同じく`Publishers`に`extension`で新たに`Publisher`を追加するやり方で`throttle(while:on:)`を追加していく。

# Publisherの作成
### ThrottleWhile structの作成

まず`Publishers`を`extension`して`Publisher`に準拠した`ThrottleWhile`という`struct`を作成する。

```Swift
extension Publishers {
    public struct ThrottleWhile<Upstream, ShouldThrottle, Context>: Publisher where Upstream: Publisher, ShouldThrottle: Publisher<Bool, Never>, Context: Combine.Scheduler {
        public typealias Output = Upstream.Output
        public typealias Failure = Upstream.Failure

        public let upstream: Upstream
        public let shouldThrottle: ShouldThrottle
        public let initialThrottle: Bool
        public let context: Context

        public init(upstream: Upstream, shouldThrottle: ShouldThrottle, initialThrottle: Bool, context: Context){
            self.upstream = upstream
            self.shouldThrottle = shouldThrottle
            self.initialThrottle = initialThrottle
            self.context = context
        }
```

プロパティは上流の`Publisher`である`upstream: Upstream`と値をスロットルするための`Publisher`である`shouldThrottle: ShouldThrottle`、そして`Publisher`はReactiveSwiftの`Property`のように初期値を持たないため、代わりの初期値となる`initialThrottle: Bool`とスケジューラーである`context: Context`だ。

Publisherのtypealiasは、このオペレーターは`map(_:)`のように値を変更するものではないため、上流のものをそのまま使う。

### `receive(subscriber:)`関数の実装

次に`Publisher`の`receive(subscriber:)`関数の実装をする。

```Swift
        public func receive<S>(subscriber: S) where S : Subscriber, Upstream.Failure == S.Failure, Upstream.Output == S.Input {
            let inner = Inner(subscriber: subscriber, upstream: self.upstream, shouldThrottle: self.shouldThrottle, initialThrottle: self.initialThrottle, context: self.context)
            subscriber.receive(subscription: inner)
        }
```

この関数は新たに`subscriber`が接続された時に呼び出されるもので、渡された`Subscrber`へ`Subscription`を渡す。
今回は`Subscription`に準拠した`Inner` classを作成し、それを渡した。

### `Inner`クラスの作成
`Inner`クラスは`Subscription`に準拠し、`throttle(while:on:)`の実際の処理を書く。
```Swift
        private final class Inner<S: Subscriber>: Subscription where S.Input == Upstream.Output, S.Failure == Upstream.Failure {
            private let subscriber: S
            private let context: Context

            private var cancellables = Set<AnyCancellable>()

            private var demand: Subscribers.Demand = .none

            private var isThrottling: Bool
            private var latestValue: Upstream.Output?

            init(subscriber: S, upstream: Upstream, shouldThrottle: ShouldThrottle, initialThrottle: Bool, context: Context) {
                self.subscriber = subscriber
                self.context = context

                isThrottling = initialThrottle

                shouldThrottle.sink { [weak self] value in
                    self?.isThrottling = value
                    if let latestValue = self?.latestValue {
                        self?.sendValue(latestValue)
                    }
                }
                .store(in: &cancellables)

                upstream.sink { [weak self] completion in
                    self?.subscriber.receive(completion: completion)
                } receiveValue: { [weak self] value in
                    self?.latestValue = value
                    self?.sendValue(value)
                }
                .store(in: &cancellables)
            }
```
まず、`subscriber`とその他`ThrottleWhile`に必要な値を`init`で受け取り、各`Publisher`を購読する。

`shouldThrottle`の購読では`isThrottling`の値を更新し`latestValue`が設定されていればそれを`sendValue(_:)`する。

`upstream`の購読では`completion`が来たら`subscriber`に`completion`を送信し、値が来た時は`latestValue`の値を更新し`sendValue(_:)`する。

`sendValue(_:)`はプライベートメソッドで、`isThrottling`が`false`かつ、`demand`が残っているとき`subscriber`に値を送信する。

```Swift
            private func sendValue(_ value: Upstream.Output) {
                guard !isThrottling, demand > 0 else { return }
                context.schedule { [weak self] in
                    guard let self else { return }
                    demand += subscriber.receive(value)
                }
            }
```

`demand`は簡単に言うと`subscriber`が値を送って欲しい回数のことであり、`Subscription`で定義されている`request(_:)`で受け取ることができる。
```Swift 
            func request(_ demand: Subscribers.Demand) {
                guard demand > 0 else { return }
                self.demand += demand
            }
```

あとは、残りの`Subscription`が準拠している`Cancellable`で定義されている`cancel()`を実装すれば完成だ。

```Swift
            func cancel() {
                cancellables.forEach { $0.cancel() }
                cancellables.removeAll()
            }
```

# オペレーターの追加
最後に、先に作成した`Publisher`を返却するオペレーターを`Publisher`に追加する。

今回は二種類のオペレーターを作成した。

一つ目は`shouldThrottle`, `initialThrottle`, `scheduler`を引数に持つオペレーターで`ThrottleWhile`に必要な値をそのまま引数にしたものだ。
```Swift
extension Publisher {
    public func throttle<T, S>(while shouldThrottle: T, initialThrottle: Bool, on scheduler: S) -> Publishers.ThrottleWhile<Self, T, S> where T: Publisher<Bool, Never>, S: Combine.Scheduler {
        Publishers.ThrottleWhile(upstream: self, shouldThrottle: shouldThrottle, initialThrottle: initialThrottle, context: scheduler)
    }
}
```
利用例
```Swift
subject
    .throttle(while: throttle, initialThrottle: false,  on: ImmediateScheduler.shared)
    .sink {
        print("value: \($0)")
    }
    .store(in: &cancellables)
```

二つ目は`shouldThrottle`を`CurrentValueSubject`に限定することで、`initialThrottle`の設定を不要にしたものだ。

これを使えばReactiveSwiftの`throttle(while:on:)`とほぼ同じ使い方ができる。

```Swift
extension Publisher {
    public func throttle<T, S>(while shouldThrottle: T, on scheduler: S) -> Publishers.ThrottleWhile<Self, T, S> where T: CurrentValueSubject<Bool, Never>, S: Combine.Scheduler {
        Publishers.ThrottleWhile(upstream: self, shouldThrottle: shouldThrottle, initialThrottle: shouldThrottle.value, context: scheduler)
    }
}
```
利用例
```Swift
subject
    .throttle(while: throttle, on: ImmediateScheduler.shared)
    .sink {
        print("value: \($0)")
    }
    .store(in: &cancellables)
```
ReactiveSwiftの`throttle(while:on:)`の例

```Swift
producer
    .throttle(while: throttle, on: ImmediateScheduler())
    .startWithValues {
        print("value: \($0)")
   }
```

# コード全文

https://gist.github.com/hayason00/7677a59703a9ae73ac9892fadafdfbad

# 後書き
今回は、CombineでReactiveSwiftのオペレーターを再現してみた。

一般的にオペレーターの追加は既存のオペレーターを利用する形で実装する場合が多いが、今回は公式の実装に倣う形で作成することで、Combineの内部構造について深く理解することができた。

一見するとCombineはプロトコル指向で取っ付きにくいと感じるかもしれないが、実際には拡張性が高く、柔軟に対応できることが分かる。

もし現在Combineを何となく使っていて、内部構造を詳しく知りたければ、オペレーターの作成から始めてみるのが良いだろう。

ところで、既にCombineはObservationへの置き換えが始まっており、ストリーム的な処理もasync-algorithmsで置き換えられるようになっている。
未だSwift6のSendableに対応していないのも含め、Combineが非推奨になる日は近いかもしれない。

https://qiita.com/k_awoki/items/a71d8e833f99fedda7c5

https://github.com/apple/swift-async-algorithms