C# でリアクティブプログラミングを自作してみる：Select と Where オペレーター

# C# でリアクティブプログラミングを自作してみる：Select と Where オペレーター

こんにちは、@studio_meowtoon です。今回は、WSL の Ubuntu 環境の C# でリアクティブプログラミングの機能を自作する方法を紹介します。  
![csharp_on_ubuntu.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3206449/23c7c71f-e5f1-3bb8-45b6-60ad19e41238.png)

:::note alert
C# には本来[厳格なコーディング規則](https://learn.microsoft.com/ja-jp/dotnet/csharp/fundamentals/coding-style/coding-conventions)がありますが、この記事では可読性のために、一部規則に沿わない表記方法を使用しています。ご注意ください。
:::

## 開発環境

+ Windows 11 Home 22H2 を使用しています。  

:::note info
[WSL](https://learn.microsoft.com/ja-jp/windows/wsl/about) の Ubuntu を操作していきますので macOS の方も参考にして頂けます。  
:::

<details>

WSL (Microsoft Store アプリ版) [※ こちらの関連記事からインストール方法をご確認いただけます](https://qiita.com/studio_meowtoon/items/2aaa6579cb8e4fa1d874)

```sh
> wsl --version
WSL バージョン: 1.0.3.0
カーネル バージョン: 5.15.79.1
WSLg バージョン: 1.0.47
```

Ubuntu [※ こちらの関連記事からインストール方法をご確認いただけます](https://qiita.com/studio_meowtoon/items/2aaa6579cb8e4fa1d874)

```sh
$ lsb_release -a
No LSB modules are available.
Distributor ID: Ubuntu
Description:    Ubuntu 22.04.1 LTS
Release:        22.04
```

.NET SDK [※ こちらの関連記事からインストール方法をご確認いただけます](https://qiita.com/studio_meowtoon/items/d68031c91462f6808dff)

```sh
$ dotnet --list-sdks
7.0.202 [/usr/share/dotnet/sdk]
$ dotnet --version
7.0.202
```

</details>

## リアクティブプログラミングを自作

### リアクティブプログラミングとは？

:::note info
リアクティブプログラミングは、CPUのマルチコア化やクラウドの活用が進む中で、 **非同期通信に基づいたアプリケーションの設計に適している**ため、 これからのソフトウェア開発における重要な技術として注目されています。
:::

#### リアクティブプログラミングはこれまでの手法と何が違うのか？

以下の疑似コードを見てください。

```sh
a = 1
b = 2 * a
a = 2
print b
```

これまでの**命令型プログラミングの観点**では以下の出力を得ます。

```sh
a = 1
b = 2 * a
a = 2
print b
```

```sh
出力結果: 2
```

対して、**リアクティブプログラミングの観点**では以下の出力を得ます。

```sh
a = 1
b = 2 * a
a = 2
print b
```

```sh
出力結果: 4
```

#### リアクティブプログラミングの観点とは？

|リアクティブプログラミングの観点では|
|:--|
|2行目の「b = 2 * a」というコードは「b を a の2倍として定義する」という意味で解釈されます。|
|つまり、a と b の関係性を定義した後は、「 a への代入」というイベントへのリアクションとして 「 b の再計算」を常にバックグラウンドで行います。|
|このようなパラダイムのことを**リアクティブプログラミング**と呼びます。|

#### リアクティブプログラミングのメリット  
これまでの命令型プログラミングでは、特に非同期の並行処理において様々な問題がありました。例えば次のような問題です。

|命令型プログラミングの非同期の並行処理における問題点|
|:--|
|コードが複雑化する。|
|命令の実行順序を管理することができない。|
|幾重にもネストされたコールバック。(いわゆるコールバック地獄)|

リアクティブプログラミングを活用すると、**今までとは異なるプログラミングパラダイム**で上記の問題を解決するコードを記述することが可能になります。

#### とは言われるものの…🤔

:::note info
一般的には先ほどのように**非同期処理と絡めて**リアクティブプログラミングの利点が紹介されるケースが多いですが、学習当初、個人的には全然実感がわきませんでした😭 そこで、この記事では**自分が納得いく方法**で説明を試みたいと思います😋
:::

#### 今までとは異なるプログラミングパラダイムって何だろう？🤔
これは個人的な見解で他のWEBサイトなどの受け売りではないのですが、[Rx](https://docs.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh242985(v=vs.103)?redirectedfrom=MSDN)系のライブラリを利用すると

+ for, foreach などの**ループ処理**を書かないようになる
+ if, else if などの**分岐処理**を書かないようになる

というコードに落ち着くかと思います。何故そうなるかを、これから**自作のRx実装**を試みながら説明していきたいと思います。

### リアクティブプログラミングを実現する方法は？

:::note info
一般的にリアクティブプログラミングの為に使用されるライブラリは、その根幹に**オブザーバーパターン**という考え方を持っています。
:::

#### GoF のオブザーバーパターン (Observer pattern)
+ オブザーバーパターンは 1995年に GoF によって提案された23のデザインパターンの一つであり、プログラムのインタフェースとしては古典的なものです。
+ **Subject(監視対象)** と **Observer(観察者)** からなるパラダイムで、利用シーンとしては、GUIの設計においてデータベースの変更を検知して表示を自動更新したい場合などが代表的です。

#### オブザーバーパターンについては
+ ここでは詳しく解説しません、別の一般的な資料を参照することをお勧めします。
+ 理由は Rx では表向きオブザーバーパターンのインターフェイスを利用しますが、どうやら内部的にはそれに準拠しない実装で実現していると推測されるからです。(※その部分は後述します)

#### まず C# でシンプルなオブザーバーパターンを実装する

:::note info
Ubuntu に C# 開発環境を構築する方法は、こちらの関連記事からご確認いただけます。
:::

https://qiita.com/studio_meowtoon/items/d68031c91462f6808dff

幸いなことに **.NET** では以下のインターフェイスが定義されています。

```mermaid
classDiagram
  class IObserver{
    +void OnNext(T value)
    +void OnError(Exception e)
    +void OnCompleted()
  }
  class IObservable{
    +IDisposable Subscribe(IObserver observer)
  }
```

+ IObserver\<T\> インターフェイス：
    + **値を受け取る側**が実装するインターフェイス
+ IObservable\<T\> インターフェイス：
    + **値を発行する側**が実装するインターフェイス

最初にシンプルな IObserver 実装クラスを作成します。

```csharp:HogeObserver.cs
public class HogeObserver : IObserver<string> {
    public void OnCompleted() {
        throw new NotImplementedException(); // ※実装省略
    }
    public void OnError(Exception error) {
        throw new NotImplementedException(); // ※実装省略
    }
    // 変更通知を受け取るメソッド
    public void OnNext(string value) {
        Console.WriteLine("OnNext: " + value); // ここではコンソールに表示します
    }
}
```

次にシンプルな IObservable 実装クラスを作成します。

```csharp:HogeObservable.cs
public class HogeObservable : IObservable<string> {
    // observer のリストを持ちます
    List<IObserver<string>> _observers = new();
    // リストに observer を追加します
    public IDisposable Subscribe(IObserver<string> observer) {
        _observers.Add(observer); 
        return null; // ※ この実装では IDisposable には対応しません
    }
    // 登録された observer に変更内容を通知する
    public void Notify(string value) {
        // ここではリストをループ処理しています
        foreach (var observer in _observers) {
            observer.OnNext(value);
        }
    }
}
```

Main メソッドから使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // Observable と Observer を使用
        HogeObservable observable = new();
        observable.Subscribe(new HogeObserver()); // 登録するだけです
        observable.Notify("Hello"); // ※ Notify() は IObservable で定義されていません
        observable.Notify("World"); // ※ Notify() は IObservable で定義されていません
    }
}
```

コマンドラインで実行します。

```sh
$ dotnet run
OnNext: Hello
OnNext: World
```

:::note info
ここまでの作業で、C# で**オブザーバーパターン**を**実装**することができました。
:::

しかし、問題点があると思います。

:::note alert
**実行するメソッド(※例えば Notify)** がインターフェイスで定義されてない！
:::

```mermaid
classDiagram
  IObserver <|.. HogeObserver 
  IObservable <|.. HogeObservable
  HogeObservable o-- IObserver
  class IObserver{
    +void OnNext(T value)
    +void OnError(Exception e)
    +void OnCompleted()
  }
  class IObservable{
    +IDisposable Subscribe(IObserver observer)
  }
  class HogeObserver{
    +void OnNext(T value)
  }
  class HogeObservable{
    +IDisposable Subscribe(IObserver observer)
    +void Notify()
  }
```

#### どうするか？🤔 ⇒ 仲介役を使用して実装します！😋

幸いなことに **System.Reactive(Rx)** パッケージで以下のインターフェイスが定義されています。

```mermaid
classDiagram
  class ISubject{
    +void OnNext(T value)
    +void OnError(Exception e)
    +void OnCompleted()
    +IDisposable Subscribe(IObserver observer)
  }
```

+ ISubject\<T\>インターフェイス：
    + **値を受け取る側** が実装するインターフェイス
    + **値を発行する側** が実装するインターフェイス

どちらの操作も持つ **ISubject** インタフェースを使います。

:::note warn
System.Reactive パッケージを NuGet でインストールします。
:::

```sh
$ dotnet add package System.Reactive
```

シンプルな ISubject 実装クラスを作成してみます。

```csharp:HogeSubject.cs
class HogeSubject : ISubject<string> {
    // observer のリストを持ちます
    List<IObserver<string>> _observers = new();
    public void OnCompleted() {
        throw new NotImplementedException(); // ※実装省略
    }
    public void OnError(Exception error) {
        throw new NotImplementedException(); // ※実装省略
    }
    // ここが呼ばれて初めて処理が走る
    public void OnNext(string value) {
        foreach (var observer in _observers) {
            observer.OnNext(value);
        }
    }
    // ここでは observer が登録されるだけ
    public IDisposable Subscribe(IObserver<string> observer) {
        _observers.Add(observer);
        return null; // ※ IDisposable には対応してない
    }
}
```

Main メソッドから使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // Subject と Observer を使用
        HogeSubject subject = new();
        subject.Subscribe(new HogeObserver());
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

コマンドラインで実行します。

```sh
$ dotnet run
OnNext: Hello
OnNext: World
```

:::note info
ここまでの作業で、Subject オブジェクトの OnNext メソッドで処理が実行されるようになりました。
:::

#### どういうこと？🤔

つまり Subject の OnNext() を**呼ぶと** Observer の OnNext() が**呼ばれ**ます。
 
```csharp
public class HogeObserver : IObserver<string> {
    // ※省略
    public void OnNext(string value) {
        Console.WriteLine("OnNext: " + value); // ここが呼ばれます
    }
}
```
これは Subject から Observer にイベント通知されると表現されます。イベント通知という言葉で**難しく説明され**ますが、単純に Subject が自分で Observer のリストを持っていて、それらをループ処理して Observer のメソッドを呼んでいます。

また、以下のクラス図のように **HogeObserver** クラスと **HogeSubject** クラスは**直接依存しません**。

```mermaid
classDiagram
  IObserver <|.. HogeObserver
  IObserver o-- HogeSubject
  IObserver <|.. ISubject 
  IObservable <|.. ISubject
  ISubject <|.. HogeSubject
  class IObserver{
    +void OnNext(T value)
    +void OnError(Exception e)
    +void OnCompleted()
  }
  class IObservable{
    +IDisposable Subscribe(IObserver observer)
  }
  class ISubject{
    +void OnNext(T value)
    +void OnError(Exception e)
    +void OnCompleted()
    +IDisposable Subscribe(IObserver observer)
  }
  class HogeObserver{
    +void OnNext(T value)
  }
  class HogeSubject{
    +IDisposable Subscribe(IObserver observer)
    +void OnNext(T value)
  }
```

### ここまでのまとめ
+ Subject が Observer を呼び出す。
    + Pull 型で情報**を通知し**ます。
+ Observer は Subject から呼ばれる。
    + Push 型で情報**が通知され**ます。

:::note info
ここまでの作業で、Subject オブジェクトの OnNext() メソッドを呼べば、Observer オブジェクトの処理が走るところまで実装出来ました😋
:::

### ここから一気に Rx の実装に近づけます！😤

:::note alert
**問題点：**
IObserver インタフェースを実装するのめんどくさくないですか？😫
:::

:::note info
**解決策：**
それ**ラムダ式**で出来ます！😋
:::

それでは C# のラムダ式を使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // Subject とラムダ式を使用
        HogeSubject subject = new();
        subject.Subscribe(x => Console.WriteLine("OnNext: " + x)); // ラムダ式
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

コマンドラインで実行します。

```sh
$ dotnet run
OnNext: Hello
OnNext: World
```

:::note info
ISubject の実装だけでここまで出来ました😋
:::

#### 再びどういうこと？😭

Rx のライブラリで C# の拡張メソッドを使用した糖衣構文が用意されています。

```csharp
public static class ObservableExtensions {
    // ※省略
    public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext);
    // ※省略
}
```

+ **Action<T> デリゲート**に**ラムダ式**を渡すことで実現しています。
+ 実際には内部的に IObserver の実装インスタンスを作成して onNext() で Action デリゲートが実行されます。

※ラムダ式については、こちらの以前の関連記事を参照頂けます。

https://qiita.com/studio_meowtoon/items/4ac1642a546b2e34d18f

#### ここまでのまとめ

|内容|
|:--|
|Rx 系でよく見る subject.Subscribe(x =>  x への処理)) は、C# では内部的に IObserver の実装インスタンスが作成されています。|
|subject.Subscribe() に渡すデリゲートは、IObserver の文脈で onNext() として呼ばれます。|
|Subscribe() にラムダ式を記述しただけでは処理は走りません、別途 onNext() されてはじめて処理が走ります。|

#### ちょっとまって😟

まだ全然 Rx っぽくないじゃないか？😕
 
## Rx(リアクティブエクステンション)っぽいコードとは？🤔

ここまでの内容は非常に長～い前振りで、リアクティブプログラミングとは、一般的に以下のような操作が行われるコードだと思います。

```csharp
// 処理の定義
HogeSubject subject = new();
subject.Where(x => x.Contains("H")) // "H"が含まれていたら
.Select(x => x + "ですねん") // "ですねん" を追加する
// Subscribe(※これまでの説明により、ここでは登録されるだけ)
.Subscribe(x => Console.WriteLine("OnNext: " + x));
// 処理の実行
subject.OnNext("Hello");
subject.OnNext("World");
```
```sh
$ dotnet run
OnNext: Helloですねん
```

とてもシンプルな例ですが、上記の動作を実装してみます。

|実装するオペレーター|内容|
|:--|:--|
|xWhere()|フィルタリング系オペレーター|
|xSelect()|メッセージ変換系オペレーター|
|xSubscribe()|ついでに独自実装してみます|

まずデリゲートを受ける Observer を実装します。

```csharp:AnonymousObserver.cs
// デリゲートを処理出来る IObserver の機能を提供します
public class AnonymousObserver<T> : IObserver<T> {
    Action<T> _onNext = null;
    public AnonymousObserver(Action<T> onNext) { // 引数がデリゲートである
        _onNext = onNext;
    }
    public void OnCompleted() {
        throw new NotImplementedException(); // ※実装省略
    }
    public void OnError(Exception error) {
        throw new NotImplementedException(); // ※実装省略
    }
    public void OnNext(T value) {
        _onNext(value); // デリゲートが実行されます
    }
}
```

次にデリゲートを登録できる Subscribe() を**拡張メソッド**として実装します。
※上記 AnonymousObserver クラスを使用して独自実装します。

```csharp:Extensions.cs
// Subscribe() をシンプルに実装します
public static IDisposable xSubscribe<T>(this IObservable<T> self, Action<T> onNext) {
    // IObserver の実装を生成します
    var disposable = self.Subscribe(new AnonymousObserver<T>(onNext));
    return disposable;
}
```

Main メソッドから使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // Subject とラムダ式を使用
        HogeSubject subject = new();
        subject.xSubscribe(x => Console.WriteLine("OnNext: " + x)); // ラムダ式
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

```sh
$ dotnet run
OnNext: Hello
OnNext: World
```

:::note info
Microsoft 謹製の Rx ライブラリと同じようにラムダ式で記述できました😋
:::

次にデリゲートを受ける Observable を実装します。

```csharp:AnonymousObservable.cs
// デリゲートを処理出来る IObservable の機能を提供します
public class AnonymousObservable<T> : IObservable<T> {
    Func<IObserver<T>, IDisposable> _subscribe;
    // 引数がデリゲートである
    public AnonymousObservable(Func<IObserver<T>, IDisposable> subscribe) { 
        _subscribe = subscribe;
    }
    public IDisposable Subscribe(IObserver<T> observer) {
        // デリゲートが実行されます ※ネストされた Observable を持ちます
        var disposable = _subscribe(observer); 
        return disposable;
    }
}
```

デリゲートを登録できる Where() オペレーターを実装します。

```csharp:Extensions.cs
// Where() オペレーターをシンプルに実装します
public static IObservable<T> xWhere<T>(this IObservable<T> self, Func<T, bool> predicate) {
    var observable = new AnonymousObservable<T>(observer => {
        var disposable = self.Subscribe(new AnonymousObserver<T>(value => {
            // ここが OnNext される内容
            // 条件に一致したときだけ observer に値を流します
            if (predicate(value)) {
                observer.OnNext(value);
            }
        }));
        return disposable;
    });
    return observable; // self ではなく別の observable を返す
}
```

Main メソッドから使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // xWhere を使用
        HogeSubject subject = new();
        subject.xWhere(x => x.Contains("H")) // "H" が含まれる場合のみ
        .Subscribe(x => Console.WriteLine("OnNext: " + x));
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

```sh
$ dotnet run
OnNext: Hello
```

:::note info
ここまでの作業で、Rx の Where オペレーター相当の機能を実装できました😋
:::

さらにデリゲートを登録できる Select() オペレーターを実装します。

```csharp:Extensions.cs
// Select() オペレーターをシンプルに実装します
public static IObservable<TResult> xSelect<T, TResult>(this IObservable<T> self, Func<T, TResult> selector) {
    var observable = new AnonymousObservable<TResult>(observer => {
        var disposable = self.Subscribe(new AnonymousObserver<T>(value => {
            // ここが OnNext される内容
            // 渡ってきた値を加工して observer にその値を流します
            TResult result = selector(value); 
            observer.OnNext(result);
        }));
        return disposable;
    });
    return observable; // self ではなく別の observable を返す
}
```

Main メソッドから使用してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // xSelect を使用
        HogeSubject subject = new();
        subject.xSelect(x => x + "ですねん") // "ですねん" を追加する
        .Subscribe(x => Console.WriteLine("OnNext: " + x));
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

```sh
$ dotnet run
OnNext: Helloですねん
OnNext: Worldですねん
```

:::note info
ここまでの作業で、Rx の Select オペレーター相当の機能を実装できました😋
:::

さらに Where と Select オペレーターを組み合わせて実行してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // xWhere と xSelect を使用
        HogeSubject subject = new();
        subject.xWhere(x => x.Contains("H")) // "H" が含まれる場合のみ
        .xSelect(x => x + "ですねん") // "ですねん" を追加する
        .xSubscribe(x => Console.WriteLine("OnNext: " + x));
        subject.OnNext("Hello");
        subject.OnNext("World");
    }
}
```

```sh
$ dotnet run
OnNext: Helloですねん
```

:::note info
ここまでの作業で、Rx の Where、Select オペレーター相当の機能を実装できました😋
:::

#### 簡易 Rx 実装の完成です🎉🎉
 
#### もちろんこの実装はいろいろ足りません😭😭
+ しかしながら、Rx(リアクティブエクステンション)の実装が内部で何をしているのかが、何となくわかって頂けたのではないでしょうか？
+ 今回は C# の実装を参考にしたので他の言語では異なる実装になると思われますが、エッセンスは共通だと思います。

### ここまでのまとめ

さて、序盤で書いた**今までとは異なるプログラミングパラダイム**って？🤔

+ for, foreach などのループ処理を書かない。
    + そもそも Subject 内の実装でラップされる。
+ if, else if などの分岐処理を書かない。
    + Where などで代替えされる。

結果、上記のようになったと思いませんか？😙
 
### 今一度、命令型プログラミングのパラダイム

最後に命令型プログラミングのパラダイムで、上記と同様の処理を実装してみます。

```csharp:Program.cs
class Program {
    static void Main(string[] args) {
        // foreach や if で実装
        List<string> list = new();
        list.Add("Hello");
        list.Add("World");
        foreach (var item in list) {
            if (item.Contains("H")) { // "H" が含まれる場合のみ
                var result = item + "ですねん"; // "ですねん" を追加する
                Console.WriteLine("result: " + result);
            }
        }
    }
}
```

```sh
$ dotnet run
result: Helloですねん
```

:::note info
全く違いますよね。今感じたことがリアクティブプログラミングとは**パラダイムが異なる**という感覚だと思います。こちらは、何のインターフェイス、クラス、拡張メソッドも使用していない為、非常にシンプルです。しかし、再利用性は限りなく低い実装と言えるでしょう。
:::

## まとめ
+ 命令型プログラミングとは異なる**リアクティブプログラミングの実装**を試すことができました。
+ 今回の説明では Rx (リアクティブエクステンション) ≠ リアクティブプログラミングのエッセンスを紹介したので **Subject.OnNext()** を自分で呼びました。
+ しかし、例えば実際の Unity & UniRx などを使用するゲームプログラミングでは、自分で **OnNext()** メソッドを呼び出すケースはほぼありません。
+ **OnNext()** メソッドは、Rx ライブラリが提供する **Observable ストリーム**を管理するライブラリ側で良きタイミングで呼び出されています。
+ この **OnNext() っていったい誰が呼び出してるんだろう？** という点が Rx 系で一番混乱する部分だと思いました。

どうでしたか？ Window 11 の WSL Ubuntu に、C# / .NET の開発環境を手軽に構築することができます。ぜひお試しください。今後も .NET の開発環境などを紹介していきますので、ぜひお楽しみにしてください。