Mu による純粋関数型RPC

いろんな純粋関数型 Scala ライブラリを作っている [Higher Kindness](https://higherkindness.io/) の RPC フレームワーク、 [Mu](https://github.com/higherkindness/mu) についての記事。すぐに動かせるミニマムなプロジェクトを作ってみた。

## Mu とはどんなものか
- もとは [Freestyle](https://github.com/frees-io/freestyle) + [gRPC](https://grpc.io/) ベースの *frees-rpc* という **RPC フレームワーク**だったものを、Mu（ミュー）に改名したもの。
- **Tagless final** の「代数」として書いた抽象度の高いプロトコル（サービス定義）から、コミュニケーション層（gRPCなど）にアクセスするコードをマクロ生成する。
- Mu の意味は**マイクロサービス**の「マイクロ(*μ*)」でもあり、マイクロサービスを視野に入れたスコープの拡張も改名の意図に含まれるという。この記事では**サーバーとクライアントの RPC 通信**を主題として、その他のマイクロサービス的な機能は最後の方で少し触れる。

## 小さなサンプルプロジェクト
Mu のリポジトリにもサンプルモジュールが含まれているが、いろいろ盛り込まれているので、どれが最低限の必須要素なのか、パッと見わかりにくい。

そこで、サーバーとクライアントの疎通に必須な要素だけを残して、いろいろ試してみるための叩き台としても使えるような、ミニマムなプロジェクトを作ってみたい。仕様は次のような最小限なもの。

- **サーバ**: 名前 `name` をリクエストで受け取って、"Hello, $name!" を返す RPC エンドポイント `greet` を公開する。
- **クライアント**: `greet` に "World" を与えて、レスポンス "Hello, World!" を得る。

#### プロジェクト構成
以下のような構成にした（仕様ライブラリやバージョンを含む詳細は [build.sbt](https://github.com/yasuabe/mu_simplest_example/blob/master/build.sbt)）。

```
modules/simplest
  ├ protocol                // server | client の両方から依存されるプロトコル定義プロジェクト
  │  └ /src/main/scala
  │      └ mu_example/simplest/protocol/Protocol.scala
  ├ server                  // protocol に依存するサーバープロジェクト
  │  ├ /src/main/resources
  │  │   └ application.conf // ポートの設定
  │  └ /src/main/scala
  │      └ mu_example/simplest/server/ServerApp.scala
  └ client                  // protocol に依存するクライアントプロジェクト
     ├ /src/main/resources
     │   └ application.conf // サーバーホストとポートの設定
     └ /src/main/scala
         └ mu_example/simplest/client/ClientApp.scala
```
できたサンプルプロジェクトは[ここに置いた](https://github.com/yasuabe/mu_simplest_example)。

## プロトコル
プロトコルは下のように Scala コードで書ける。gRPC のように *.proto ファイルを書く必要は特にない。

```scala
object Protocols {
  @message
  case class Name(value: String)       // リクエスト

  @message
  case class Greeting(value: String)   // レスポンス

  @service(Protobuf)
  trait SimplestService[F[_]] {
    def greet(name: Name): F[Greeting] // ストリームなら Stream[F, Greeting]のような型
  }
}
```
サーバー/クライアント間でやり取りされるデータは、`@message` アノテーションを付けた case class で記述する。またそれらのデータ型を用いたサービス定義は、`@service` アノテーションをつけた trait で、Tagless Final の形式で記述する。

`@service` のパラメータには、`Protobuf` の他に `Avro` も指定できるが、Avro では [ストリーミングはサポートされない](https://www.47deg.com/blog/mu-rpc-defining-messages-and-services/#service-protocol-definition-1)という。また圧縮方式として`Gzip` を指定することもできる（デフォルトは `Identity` で圧縮なし）。

このプロトコルを使う側のサーバ／クライアントのコードでは、型パラメータ *F* として Cats Effect の `ConcurrentEffect` のインスタンスがあるものを指定することになる。典型的には `IO` や `Task` が使われる。

コミュニケーション形態としては、単発の Unary RPC と、ストリームの Server/Client/Bidirectional RPC があるが、ここでは Unary RPC を使うことにする。


## サーバ
以下が、`SimplestService` の振る舞いを実装したハンドラとなる。受け取った名前をログ出力して、`Greeting` にラップして計算結果としている（`*>` は Cats の apply 構文、`pure` は applicative 構文）。

```scala
def put[F[_]: Async](line: String): F[Unit] = Async[F].delay(println(line))

class SimplestServiceHandler[F[_]: ConcurrentEffect] extends SimplestService[F] {
  def greet(n: Name): F[Greeting] =
    put[F](s"Greeting to ${n.value} ...") *> Greeting(s"Hello, ${n.value}!").pure[F]
}
```
プロトコルでは *F* はただの `F[_]` だったが、ここでは `F[_]: ConcurrentEffect` と制約がつく。ただし、どの `ConcurrentEffect` かは、まだ確定させる必要はない。

次がいわゆる main、あるいは *the end of the universe* となるクラスで、ここで *F* が具体的に選択される。このサンプルでは monix の `Task` としてみた。

```scala
object ServerApp extends TaskApp {
  implicit val alg: SimplestService[Task] = new SimplestServiceHandler[Task]
  ...

  def run(args: List[String]): Task[ExitCode] = for {
    configs <- SimplestService.bindService[Task].map(AddService)
    server  <- BuildServerFromConfig[Task]("rpc.server.port", List(configs))
    _       <- put[Task](s"Server is starting ...")
    _       <- GrpcServer.server[Task](server)
  } yield ExitCode.Success
}
```
ちなみに、`run` メソッド内1行目の `bindService` は [Mu のマクロ](https://github.com/higherkindness/mu/blob/master/modules/internal/src/main/scala/higherkindness/mu/rpc/internal/serviceImpl.scala)で生成されるもので、以下のようなシグネーチャになる。上で定義したハンドラは、暗黙にこの `bindService` メソッドに渡される。

```scala
def bindService[F[_]](
  implicit
    F:       ConcurrentEffect[F],
    algebra: SimplestService[F],  // <-- ここが固有部分
    EC:      ExecutionContext
): F[ServerServiceDefinition]
```
結果となる `ServerServiceDefinition` は gRPC のサービス定義で、これと [application.conf](https://github.com/yasuabe/mu_simplest_example/blob/master/modules/simplest/server/src/main/resources/application.conf) に記述したポート番号指定を使って、GrpcServer インスタンスを生成することができる。

さらに、生成した GrpcServer インスタンスを `GrpcServer.server[Task]` に渡すと、内部で `start` メソッドが実行され gRPC サーバーが始動する。

## クライアント
クライアントは、`Resource` として与えられるサービスのプロキシを使って `greet` メソッドを実行し、結果を標準出力する。

```scala
case class SimplestClient[F[_]: ConcurrentEffect](
    service: Resource[F, SimplestService[F]]) {

  def greet(n: String): F[Unit] = put[F](s"greeting to $n") *>
    service.use(_.greet(Name(n)))
      .flatMap { g => put(s"got a greeting from server: $g") }
      .handleErrorWith {
        case e: StatusRuntimeException => // 例えばポート設定が間違ってたりするとここに
          put(s"RPC failed: ${e.getStatus} $e") *>
            ApplicativeError[F, Throwable].raiseError(e)
      }
}
```
このインスタンスを得るには、ホストとポートを記述した [application.conf](https://github.com/yasuabe/mu_simplest_example/blob/master/modules/simplest/client/src/main/resources/application.conf) に基づいてチャンネルを生成したうえで、以下のようにする。[^channel_generation]

```scala
object SimplestClient {
  def apply[F[_]: ConcurrentEffect](implicit ec: ExecutionContext): F[SimplestClient[F]] =
    ConfigForAddress[F]("rpc.client.host", "rpc.client.port") map { channel =>
      SimplestClient[F](SimplestService.client[F](channel))
    }
}
```
上記コード中の `SimpleService.client[F]` も、サーバー側で説明した `bindService` と同様にマクロで生成されるもので、以下のようなシグネーチャになる。

```scala
def client[F[_]](
  channelFor:        ChannelFor,
  channelConfigList: List[ManagedChannelConfig] = List(UsePlaintext()),
  options:           CallOptions                = CallOptions.DEFAULT
)(
  implicit F:  ConcurrentEffect[F],
           EC: ExecutionContext
): Resource[F, SimplestService[F]]   // <-- ここが固有部分
```
実行するコードは、たとえば以下のようになる。

```scala
object ClientApp extends TaskApp {
  ...
  def run(args: List[String]): Task[ExitCode] = for {
    _      <- put[Task](s"Starting client, interpreting to Future ...")
    client <- SimplestClient[Task] // コンパニオンオブジェクトの apply が使われる
    _      <- client.greet("World")
    _      <- put[Task](s"Finishing program interpretation ...")
  } yield ExitCode.Success
}
```
*F* には、サーバー側と同じ `Task` を指定したが、別に同じである必要はなく、`IO` でも問題ない。

## 実行結果
IntelliJ からでも、`sbt server/run` あるいは `sbt client/run` でもどちらでも実行できる。まずサーバーを起動して、つぎにクライアントを実行すると以下のようになる。
#### サーバ
```
Server is starting ...
Greeting to World ...
```
#### クライアント

```
11: Starting client ...
greeting to World
got a greeting from server: Greet(Hello, World!)
11: Finishing program ...
```


## 他にできること
最初に書いたとおり、このサンプルはあえて最小限なコードを目指したが、他にも提供されている機能や選択肢がある。

- 上で少し触れたように、単発の Unary 呼び出しだけではなく monix の `Observer`、fs2 の `Stream` を使ったストリーミングRPCも提供されている。
- ここでは Protocol Buffer を使ったが、Avro も指定できる（ただしストリーミングは不可）。
- [Prometheus](https://prometheus.io/) と連携したメトリクスレポーティング機能がある（[記事](http://higherkindness.io/mu/metrics-reporting)）。
- Scala で書いた プロトコル定義から IDL が生成できる（[記事](https://higherkindness.io/mu/idl-generation)）。
- Avro（現状） IDL から、Scala コードが生成できる（[記事](https://higherkindness.io/mu/idl-generation#generating-source-files-from-idl)）。
- SSL/TSL を用いて通信が暗号化できる（[記事](http://higherkindness.io/mu/ssl-tls)）
- テスティングフレームワークも提供されている（[記事](http://higherkindness.io/mu/patterns#service-testing)）。


## 所感
- Cats, Cats Effect, Monix, FS2 などと相性がいいので、普段からそれらを使っているスキルセットや環境だったりすると使いやすそう。
- 今のところ、以下のような点がマイクロサービスに役立ちそう。
  - [Remote Procedure Invocation](https://qiita.com/yasuabe2613/items/3bff44e662c922083264#remote-procedure-invocation): このフレームワークの主たる機能。かなり関数型が徹底されていて良い。
  - [Application Metrics](https://qiita.com/yasuabe2613/items/3bff44e662c922083264#application-metrics): Prometheus などとの連携が、最初から提供されている点。
- ただし、マイクロサービスといっても[取り扱っている問題領域](https://qiita.com/yasuabe2613/items/3bff44e662c922083264#問題領域のカタログとして役立つ)は実はかなり広いので、Mu だけで「マイクロサービスベースのアーキテクチャのための関数型フレームワーク[^microservice_framework]」というのはやや大げさかもしれない。
- プロトコル定義の Scala コードに、IDL生成のための付随情報をアノテーションで付与するのは、関心事が [complect](https://www.infoq.com/presentations/Simple-Made-Easy) してしまっていると見ることもできて、これはちょっと避けたい。

## 資料
- [mu github](https://github.com/higherkindness/mu)
- [mu site](http://higherkindness.io/mu/) 
- [47 Degrees: Introducing Mu: A purely functional library for building microservices](https://www.47deg.com/blog/introducing-the-mu-open-source-library/)
- [47 Degrees: Mu-RPC: defining messages and services](https://www.47deg.com/blog/mu-rpc-defining-messages-and-services/)

[^channel_generation]: Mu のサンプルでも、また 47 Degrees の Mu 記事でも、`unsafeRunSync` で 生の`channel` オブジェクトを得ているが、このサンプルでは途中で、つまり 'the end of the world' 以前にアンセーフなコードを書くのは避けた。
[^microservice_framework]: a functional framework for building software architectures based on microservices