【Laravel】 キュー・イベント・ブロードキャストに関する補足とフロントエンドへの導入

**【2021-11-29 追記】**
**フロントエンドで Redux-Saga が採用されるケースが限定的になってきているので，フロントエンド結合部分に関してはレガシーな内容だと思って軽く読み飛ばしてください。**

# はじめに

- [キュー 5.7 Laravel](https://readouble.com/laravel/5.7/ja/queues.html)
- [イベント 5.7 Laravel](https://readouble.com/laravel/5.7/ja/events.html)
- [ブロードキャスト 5.7 Laravel](https://readouble.com/laravel/5.7/ja/broadcasting.html)

前回同様，基本はマニュアルを参照すべし。以下補足説明

# キューのアーキテクチャ

| 名称 | 説明 |
|:---|:---|
| キュー | バックグラウンドジョブが突っ込まれる場所 |
| ワーカー | バックグラウンドジョブを取り出して消化するプロセス |
| コネクション　| バックグラウンドジョブを突っ込むための入口 |
| バス | 複数のコネクションを束ねる，入口のための入口 |

- コネクションは複数のキューを持つことができる。一般的に，ジョブの種類や優先度によってキューを分岐することが多い。
- キューとワーカーは一対一で対応するとは限らず，「どのワーカーがどのキューを処理するか」ということに関して任意の組み合わせが可能である。

![キュー](https://qiita-image-store.s3.amazonaws.com/0/25060/bba19e6c-57f8-d58f-7a0a-b1646d5afb25.png)

総称として，バスを除くこれらの仕組みをまとめて「キュー」と呼ぶこともある。

# ジョブのアーキテクチャ

## ジョブをディスパッチする方法

マニュアルには

>```php
>// このジョブはデフォルトキューへ送られる
>Job::dispatch();
>
>// このジョブは"emails"キューへ送られる
>Job::dispatch()->onQueue('emails');
>```

のようなコードが書かれているが，これは Artisan コマンドで生成したジョブクラスに最初からミックスインされている [`Dispatchable`](https://github.com/laravel/framework/blob/d5d72598ec9bbc006b3535209a895e73afef4b5e/src/Illuminate/Foundation/Bus/Dispatchable.php) トレイトに実装されているメソッドである。

```php
<?php

namespace Illuminate\Foundation\Bus;

use Illuminate\Contracts\Bus\Dispatcher;

class PendingDispatch
{
    /**
     * The job.
     *
     * @var mixed
     */
    protected $job;

    /**
     * Create a new pending job dispatch.
     *
     * @param  mixed  $job
     * @return void
     */
    public function __construct($job)
    {
        $this->job = $job;
    }

    /* ... */

    /**
     * Handle the object's destruction.
     *
     * @return void
     */
    public function __destruct()
    {
        app(Dispatcher::class)->dispatch($this->job);
    }
}
```

メソッドチェーンを通じて設定を行い，チェーンが切れたタイミングでディスパッチされる。最終的にやっていることは

```php
app(\Illuminate\Contracts\Bus\Dispatcher::class)->dispatch($job);
```

だ。ちなみに `Illuminate\Contracts\Bus\Dispatcher` に対応するファサード `Illuminate\Support\Facades\Bus` が存在しているので，ファサードを使って

```php
Bus::dispatch($job);
```

と書くこともできる。

なお，ディスパッチされたジョブは，**[`\Illuminate\Contracts\Queue\ShouldQueue` を実装しているかどうかによって，直ちに実行するか，一旦キューに投入するかの分岐が行われる](https://github.com/illuminate/bus/blob/d2d1f70246232460799e0412c0ec7c71ba85d11b/Dispatcher.php#L64-L77)**。必ずバックグラウンドジョブにしたい場合は実装しておこう。

## 「ジョブ」という言葉の定義

Laravel はジョブの仕組みをレイヤー化しているが，**重複して同じ「ジョブ」という名称で呼んでいるものがある**ため，フレームワークのコードを読むと混乱しやすい。ここではレイヤー別にジョブについての説明を行う。

それぞれ，インスタンスが生成されるタイミングに注意されたい。

| 名称 | レイヤー |予約時のインスタンス生成 | 実行時のインスタンス生成 |
|:--|:--:|:--:|:--:|
| アプリケーションジョブ |高| ✔ | ✔<br>(デシリアライズ) |
| ラッパージョブ |中| | ✔ |
| ドライバジョブ |低| | ✔ |

<ins>A. キューに投入されるまでの大まかな流れ</ins>

1. **アプリケーションジョブ**のインスタンスを生成し，ディスパッチ処理を開始。
2. アプリケーションジョブが [`serialize()`](http://php.net/manual/ja/function.serialize.php) 関数でシリアライズされる。
3. 更にシリアルを固定の**ラッパージョブ名 `CallQueuedHandler@call`**とともにJSONとしてラップし，これがペイロードとなってキューに投入される。

![enqueue](https://qiita-image-store.s3.amazonaws.com/0/25060/79aeb93c-8573-8de0-131f-cb40d00cab9e.png)

<ins>B. キューから取り出されて実行されるされるまでの大まかな流れ</ins>

1. キューからペイロードが取り出され， **ドライバジョブ**のインスタンスが生成される。
2. **ドライバジョブ**の**`fire()`**メソッドが実行される。
3. **ラッパージョブ`CallQueuedHandler`**のインスタンスが生成され， アプリケーションジョブのシリアルを引数として**`call()`**メソッドが実行される。
4. [`unserialize()`](http://php.net/manual/ja/function.unserialize.php) 関数でアプリケーションジョブがデシリアライズされる。
4. **アプリケーションジョブ**の**`handle()`**メソッドが実行される。

![dequeue](https://qiita-image-store.s3.amazonaws.com/0/25060/069415a0-bbeb-ad55-4488-d2d4e28be6c7.png)

### アプリケーションジョブ

`./artisan make:job` で作る，通常 `App\Jobs\` 名前空間に配置されるジョブのことである。マニュアルが「ジョブ」と言ったら基本的にこれのことを指している。

>```php
><?php
>
>namespace App\Jobs;
>
>/* ... */
>
>class ProcessPodcast implements ShouldQueue
>{
>    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
>
>    protected $podcast;
>
>    /**
>     * 新しいジョブインスタンスの生成
>     *
>     * @param  Podcast  $podcast
>     * @return void
>     */
>    public function __construct(Podcast $podcast)
>    {
>        $this->podcast = $podcast;
>    }
>
>    /**
>     * ジョブの実行
>     *
>     * @param  AudioProcessor  $processor
>     * @return void
>     */
>    public function handle(AudioProcessor $processor)
>    {
>        // アップロード済みポッドキャストの処理…
>    }
>}
>```

参考程度に，トレイトの解説も行っておく。ちなみにこれらのトレイトはすべて任意使用であって，必須ではない。

| トレイト | 所属サービス |役割 |
|:----|:---|:----|
| [`Dispatchable`](https://github.com/laravel/framework/blob/d5d72598ec9bbc006b3535209a895e73afef4b5e/src/Illuminate/Foundation/Bus/Dispatchable.php) | Bus | ディスパッチ処理をインスタンスメソッドから呼べるようにする |
| [`InteractsWithQueue`](https://github.com/illuminate/queue/blob/da95d672212f2f30cf66595ec9e256c0f08ed2f7/InteractsWithQueue.php) | Queue | ドライバジョブにアプリケーションジョブからアクセスできるようにする |
| [`Queueable`](https://github.com/illuminate/bus/blob/f032e8a42369468d4ed39a4d1cbf6109d9a04a53/Queueable.php) | Bus | 使用するコネクション・キューをデフォルトから変更できるようにする |
| [`SerializesModels`](https://github.com/illuminate/queue/blob/da95d672212f2f30cf66595ec9e256c0f08ed2f7/SerializesModels.php) | Queue | Eloquent モデルオブジェクト自体をシリアライズせず，主キーのみをシリアライズし，デシリアライズされるときに再度クエリを投げて取得する<br>**（これが無いと，デシリアライズされたオブジェクトは読み取り専用になってしまう）** |

### ラッパージョブ

基本的に **[`Illuminate\Queue\CallQueuedHandler::call()`](https://github.com/illuminate/queue/blob/da95d672212f2f30cf66595ec9e256c0f08ed2f7/CallQueuedHandler.php)** で固定。以下のような処理を行うために，**アプリケーションジョブをラップする**形で使用される。

- アプリケーションジョブのシリアライズ・デシリアライズ
- [`InteractsWithQueue`](https://github.com/illuminate/queue/blob/da95d672212f2f30cf66595ec9e256c0f08ed2f7/InteractsWithQueue.php)を使用しているアプリケーションジョブにドライバジョブをセットする
- チェインされたアプリケーションジョブの継続処理
- 完了または失敗したドライバジョブの削除処理
- [`SerializesModels`](https://github.com/illuminate/queue/blob/da95d672212f2f30cf66595ec9e256c0f08ed2f7/SerializesModels.php) を実装したアプリケーションジョブのデシリアライズで [`ModelNotFoundException`](https://github.com/illuminate/database/blob/b44d6494c24c93a16205fcf9b8db171e9397e48d/Eloquent/ModelNotFoundException.php) が発生したとき， 以下のデフォルトプロパティ宣言があれば，ジョブを失敗と見なさずに直ちに削除する

```php
public $deleteWhenMissingModels = true;
```

### ドライバジョブ

ドライバジョブは， [`Job` 契約](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Queue/Job.php) を実装している。参考程度に，ドライバジョブとしてフレームワークによって定義されている例を挙げておく。

- [queue/RedisJob.php at master · illuminate/queue](https://github.com/illuminate/queue/blob/master/Jobs/RedisJob.php)
- [queue/DatabaseJob.php at master · illuminate/queue](https://github.com/illuminate/queue/blob/master/Jobs/DatabaseJob.php)

なおこれらは，[基底抽象クラス `Job`](https://github.com/illuminate/queue/blob/master/Jobs/Job.php)を利用して，契約を利用しつつもコードのコピペを避けるように工夫されている。

# イベントのアーキテクチャ

イベントは，一見ジョブと似ているようにも思えるが，**発生したときの処理を外部に移譲する**という点で決定的に異なる。その処理を行うのはリスナーだ。どちらかといえば，リスナーのほうが素のジョブに近いだろう。

以下では，リスナーに [`ShouldQueue`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Queue/ShouldQueue.php) が実装されている場合の図解を示す。

<br><ins>A. リスナーがキューに投入されるまでの大まかな流れ</ins>

1. イベントが発生する。
2. 呼び出すリスナーの情報を引数として **[`CallQueuedListener`](https://github.com/illuminate/events/blob/eaf87f7a83b3a199f855379d5ccc1c63dc88574b/CallQueuedListener.php)** クラスのインスタンスが生成され，以降アプリケーションジョブとして処理される。 この際， `"displayName"` にリスナーのクラス名が自動的に設定され，デバッグ情報としてワーカーのログに表示されるようになる。

![enqueue](https://qiita-image-store.s3.amazonaws.com/0/25060/e38d0bcd-6154-85f3-b281-4e2d1d62cba1.png)

<ins>B. キューから取り出されて実行されるされるまでの大まかな流れ</ins>

1. アプリケーションジョブ **[`CallQueuedListener`](https://github.com/illuminate/events/blob/eaf87f7a83b3a199f855379d5ccc1c63dc88574b/CallQueuedListener.php)** のインスタンスが復元される。
2. **`handle()`** メソッドが実行される。
3. 元のリスナーのインスタンスが生成され，その `handle()` メソッドが実行される。

![dequeue](https://qiita-image-store.s3.amazonaws.com/0/25060/e8bb05f8-e7e9-adb6-962f-65e32dc18ff0.png)

# ブロードキャストのアーキテクチャ

## ブロードキャストなのにキューが必要な理由？

通常の [`ShouldBroadcast`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Broadcasting/ShouldBroadcast.php) 契約をイベントに実装している場合，イベントをブロードキャストするためには

1. キュードライバ
2. ブロードキャストドライバ

の2つが必要だ。どうしてそうなっているのかを考えてみよう。

以下にそれぞれのドライバを列挙する。なお，処理が同期的に行われてキューの意味が無いものであったり，ログを取るだけもしくは何もしない，といった選択肢は除外している。

<br><ins>キュードライバ一覧</ins>

- database
- beanstalkd
- sqs
- redis

<ins>ブロードキャストドライバ一覧</ins>

- pusher
- redis

この中に1つ明らかに浮いているものがある。それは Pusher だ。 **Pusher はAPI経由で外部サービスとして利用するしかないため，通信のオーバーヘッドがそれなりに発生する。**SQS も AWS のサービスではあるが，AWS でインフラを構築している場合しか使用しないので， 外部サービスではないと考えられる。

もし Pusher をブロードキャストドライバに選択した場合，ブロードキャスト処理を行うだけでリクエストに対する応答が遅くなってしまう。これを克服するためにはブロードキャスト処理をバックグラウンドで行う必要があり，そのためのキュー投入処理があるのだ。

逆に言えば，ブロードキャストドライバに Redis を利用している場合，IO が非常に軽いため，キューに投入する処理は完全に無駄になってしまう。（もしキュードライバにも Redis を利用していたら， Redis と無駄に2回通信することになってしまう）

**ブロードキャストドライバに Redis を採用した場合， [`ShouldBroadcast`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Broadcasting/ShouldBroadcast.php) は一切使用せず，すべて [`ShouldBroadcastNow`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Broadcasting/ShouldBroadcastNow.php) に統一すべきだ。**但し，イベントの `broadcastWith` メソッドの処理が極端に重くなったりする場合はこの限りではない。

### [`ShouldBroadcast`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Broadcasting/ShouldBroadcast.php) を実装したイベント

ブロードキャストドライバに Pusher を使用する場合の図解を示す。

![enqueue](https://qiita-image-store.s3.amazonaws.com/0/25060/0820495f-909a-345f-731e-b11526bde86c.png)
![dequeue](https://qiita-image-store.s3.amazonaws.com/0/25060/0d7a4cdc-ae57-a4a7-199d-9abdc80bb7ae.png)

いったんキューで寝かせて，出番が来たら Pusher に丸投げ。

### [`ShouldBroadcastNow`](https://github.com/illuminate/contracts/blob/963431ba09c41e810b0f0ca303ba9368671ba5dd/Broadcasting/ShouldBroadcastNow.php) を実装したイベント

ブロードキャストドライバに Redis + Laravel Echo Server を使用する場合の図解を示す。

![download-4.png](https://qiita-image-store.s3.amazonaws.com/0/25060/0ed7aeaa-eb27-293d-5b44-7ca12bd31fd0.png)

Redis に直接 `PUBLISH` して， Laravel Echo Server が `SUBSCRIBE` する形になる。

## 特定イベントに関して `toOthers()` メソッドを常に呼ぶ

> しかし、`broadcast`関数には、ブロードキャストの受取人から現在のユーザーを除外できる、`toOthers`メソッドが用意されています。
> 
> ```php
> broadcast(new ShippingStatusUpdated($update))->toOthers();
> ```

特定のイベントに関して常に `toOthers()` を利用したい場合，イベントのコンストラクタで [`dontBroadcastToCurrentUser()`](https://github.com/illuminate/broadcasting/blob/1ffe1a548b1f644eeefb0d42b854ddca12cd5b06/PendingBroadcast.php#L36-L48) をコールするとよい。 以下に例を示す。

```php:イベントの実装例
<?php

declare(strict_types=1);

namespace App\Events;

use App\Post;
use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Contracts\Broadcasting\ShouldBroadcastNow;
use Illuminate\Queue\SerializesModels;

class BroadcastPostCreated implements ShouldBroadcastNow
{
    use SerializesModels, InteractsWithSockets;

    /**
     * @var Post
     */
    public $post;

    public function __construct(Post $post)
    {
        $this->dontBroadcastToCurrentUser();
        $this->post = $post;
    }

    public function broadcastWith(): array
    {
        return $this->post->toArray();
    }

    public function broadcastAs(): string
    {
        return 'post.created';
    }

    public function broadcastOn(): array
    {
        /* ... */
    }
}
```

`toOthers()` はこれを呼んでいる。

```php
/**
 * Broadcast the event to everyone except the current user.
 *
 * @return $this
 */
public function toOthers()
{
    if (method_exists($this->event, 'dontBroadcastToCurrentUser')) {
        $this->event->dontBroadcastToCurrentUser();
    }

    return $this;
}
```

このメソッドは [`InteractsWithSockets`](https://github.com/illuminate/broadcasting/blob/1ffe1a548b1f644eeefb0d42b854ddca12cd5b06/InteractsWithSockets.php) トレイトによって提供される。

```php
/**
 * Exclude the current user from receiving the broadcast.
 *
 * @return $this
 */
public function dontBroadcastToCurrentUser()
{
    $this->socket = Broadcast::socket();
    return $this;
}
```

`Broadcast::socket()` の実態は [`BroadcastManager`](https://github.com/illuminate/broadcasting/blob/1ffe1a548b1f644eeefb0d42b854ddca12cd5b06/BroadcastManager.php) に実装されている。

```php
/**
 * Get the socket ID for the given request.
 *
 * @param  \Illuminate\Http\Request|null  $request
 * @return string|null
 */
public function socket($request = null)
{
    if (! $request && ! $this->app->bound('request')) {
        return;
    }

    $request = $request ?: $this->app['request'];

    return $request->header('X-Socket-ID');
}
```

## Laravel Echo を用いたフロントエンドとのつなぎ込み

PHP側からある特定のイベントがブロードキャストされたとき，フロントエンド側で対応する Redux のアクションをディスパッチさせることを考える。

### 接続状態のコントロール

[`Echo`] は生の [`WebSocket`] 同様，インスタンスを生成した瞬間に接続試行されるため， Redux などで状態遷移を管理している場合には扱いにくい。ラップした上で， `connect()` メソッドを呼んで接続するようにすると扱いやすい。以下に Socket.IO を使用する場合の実装例を示す。

```js:src/api/echo.js
import io from 'socket.io-client'
import Echo from 'laravel-echo'
/* ... */

class EchoContainer {

  connection = null

  connecting = () => {
    return !!this.connection
  }

  connect = () => {
    this.connection = new Echo({
      broadcaster: 'socket.io',
      client: io,
      host: 'https://example.com',
    })
    return new Promise((resolve, reject) => {
      if (!this.connecting()) {
        throw new Error('socket already disconnected')
      }
      this.connection.connector.socket.on('connect', resolve)
      this.connection.connector.socket.on('error', reject)
    })
  }

  disconnect = () => {
    if (this.connecting()) {
      this.connection.disconnect()
      this.connection = null
    }
  }

  /* ... */
}

export default new EchoContainer()
```

※ `this` のバインドを省略できるよう，インスタンスごとに関数オブジェクトを作る書き方をしている

### [`axios`] からのリクエストへの `X-Socket-Id` ヘッダの付与

既に紹介したように，Laravel のブロードキャストは，「現在のリクエストを送ってきた相手にはブロードキャストしない」という機能を持っているが，これに必要な `X-Socket-Id` として付与すべきIDは以下のように取り出せる。これもラップしたクラスに実装するとよい。以下に [`axios`] を利用している場合の実装例を示す。

```js:src/api/echo.js
/* ... */

class EchoContainer {

  /* ... */

  interceptor = (config) => {
    if (this.connecting() && config.method !== 'get') {
      const socketId = this.connection.connector.socketId()
      if (socketId) {
        config.headers['X-Socket-Id'] = socketId
      }
    }
    return config
  }

  /* ... */
}

export default new EchoContainer()
```

```js:src/api/app.js
import axios from 'axios'
import echo from './echo'

const app = axios.create()
app.interceptors.request.use(echo.interceptor)

/* ... */

export default app
```

基本的に GET メソッドのエンドポイントアクセスでブロードキャストが発生することは無いので，そのときのみ除外している。これは，**APIサーバがクロスオリジンである場合，独自ヘッダーが使用されるとプリフライトリクエストが飛んでしまう**のを避ける意図がある。GET アクセスで毎回通信2往復になるのは，サービス品質的に大きな問題があるだろう。

### [`redux-saga`] とのインテグレーション例

Echo（および内包される Socket.IO）を素の Event Emitter として使うことも可能ではあるが，状態遷移を扱いやすくする場合は， [`redux-saga`] の [Event Channel](https://github.com/redux-saga/redux-saga/blob/fc6a2e3d5b1aa37864a13be6db3115ad6afc1f1b/docs/advanced/Channels.md#using-the-eventchannel-factory-to-connect-to-external-events) などを利用することが合理的であると考えられる。

![redux-saga](https://qiita-image-store.s3.amazonaws.com/0/25060/67a81965-cd4c-dad7-16d9-34198c3132d3.png)

- 1つの Connection の中で， 例えば `users.<ユーザID>` といった Echo の Channel を作成できる。認証を要求するものは Private Channel を利用する。
- Echo は，1つのチャンネルに対して複数のイベントリスナーを設定できる。このイベントリスナーでイベントの種類ごとに Saga Event Channel を生成する。

以下にこのアーキテクチャを採用したコード例を示す。まずは `EchoContainer` 側から。

```js:src/api/echo.js
/* ... */
import { eventChannel } from 'redux-saga'

class EchoContainer {

  /* ... */

  createEventChannel = (echoChannelName, echoEventName, methodName = 'channel') => {
    return eventChannel((emit) => {
      if (!this.connecting()) {
        return () => {}
      }

      // イベントリスナの登録
      const channel = this.connection[methodName](echoChannelName)
      channel.on(echoEventName, (value) => emit(value))

      // イベントに関するコールバック一覧，および最後に登録したコールバックを取得
      const callbacks = channel.events[echoEventName]
      const callback = callbacks[callbacks.length - 1]

      // ディスポーザーの定義
      return () => {
        if (!this.connecting()) {
          return
        }

        // 最後に登録したコールバックを削除する
        for (let i = callbacks.length - 1; i >= 0; --i) {
          if (callbacks[i] === callback) {
            channel.socket.removeListener(echoEventName, callback)
            callbacks.splice(i, 1)
          }
        }

        // コールバック一覧が空になったらチャンネルを破棄する
        if (callbacks.length === 0) {
          this.connection.leave(echoChannelName)
        }
      }
    })
  }

  /* ... */
}

export default new EchoContainer()
```

公式マニュアルを見ると [`listen()`](https://github.com/laravel/echo/blob/f38bf48414d4553ba93cc6e976dfb8be1ad3f440/src/channel/socketio-channel.ts#L74) でイベントリスナーを設定するように説明されているが，これは暗黙的にチャンネル名を加工する効果がある。

```ts:src/channel/socketio-channel.ts
/**
 * Listen for an event on the channel instance.
 */
listen(event: string, callback: Function): SocketIoChannel {
    this.on(this.eventFormatter.format(event), callback);

    return this;
}
```

```ts:src/util/event-formatter.ts
/**
 * Format the given event name.
 */
format(event: string): string {
    if (event.charAt(0) === '.' || event.charAt(0) === `\\`) {
        return event.substr(1);
    } else if (this.namespace) {
        event = this.namespace + '.' + event;
    }

    return event.replace(/\./g, `\\`);
}
```

この処理は**[過去に繰り返し破壊的変更が行われており](https://github.com/laravel/echo/pull/168)，バグの温床になっている。**もしPHP側で `broadcastAs()` メソッドを定義して明示的にブロードキャストされる際のイベント名を指定している場合は，暗黙的変換に頼らない **[`on()`](https://github.com/laravel/echo/blob/f38bf48414d4553ba93cc6e976dfb8be1ad3f440/src/channel/socketio-channel.ts#L90-L102)** メソッドを直接利用するほうがよい。

また衝撃的なことに， Laravel Echo には**イベントリスナー1つだけを指定して破棄するインタフェースが用意されていない**。このためリスナー解除時の関数として， [`Array#splice`](https://developer.mozilla.org/ja/docs/Web/JavaScript/Reference/Global_Objects/Array/splice) を使用して，少々トリッキーな実装を行っている。

- 備考: [Deleting array elements in JavaScript - delete vs splice - Stack Overflow](https://stackoverflow.com/questions/500606/deleting-array-elements-in-javascript-delete-vs-splice/38803368)

続いて， Saga 側を見ていこう。細かいアクションの定義は省略するが，大まかには以下のような実装になる。

```js:src/redux/echo.js
import { call, put, fork, take, race } from 'redux-saga/effects'
import echo from '../api/echo'

export function* runLoop(channel, actionCreator, UNLISTEN = null) {
  while (true) {
    let entity, unlisten
    if (UNLISTEN) {
      ({ payload, unlisten } = yield race({ payload: take(channel), unlisten: take(UNLISTEN) }))
      if (unlisten) break
    } else {
      entity = yield take(channel)
    }
    yield put(actionCreator(payload))
  }
  channel.close()
}

// 認証無しチャンネル
export function* forkForChannel(echoChannelName, echoEventName, actionCreator, UNLISTEN = null, methodName = 'channel') {
  const channel = yield call(echo.createEventChannel, echoChannelName, echoEventName, methodName)
  yield fork(runLoop, channel, actionCreator, UNLISTEN)
}

// 認証ありチャンネル
export function* forkForPrivateChannel(echoChannelName, echoEventName, actionCreator, UNLISTEN = null) {
  yield* forkForChannel(echoEventName, echoEventName, actionCreator, UNLISTEN, 'private')
}

/* ... */
```

```js:使用例
export function *listenChannels(userId) {
  yield* forkForChannel('global', 'announcement.received', broadcastAnnouncementReceived)
  yield* forkForPrivateChannel(`users.${userId}`, 'notifications.created', broadcastNotificationCreated, UNLISTEN_USER_CHANNEL)
  yield* forkForPrivateChannel(`users.${userId}`, 'notifications.updated', broadcastNotificationUpdated, UNLISTEN_USER_CHANNEL)
  yield* forkForPrivateChannel(`users.${userId}`, 'notifications.deleted', broadcastNotificationDeleted, UNLISTEN_USER_CHANNEL)
}
```

引数は以下のようになる。

| 引数名 | 必須| 説明 |
|:---|:---|:---|
| `echoChannelName` | ✔ |Echo Channel のチャンネル名 |
| `echoEventName` | ✔ | Echo Channel の各リスナーに対応するイベント名 |
| `actionCreator` | ✔ | イベントを受信したときにディスパッチしたい Redux アクションのアクションクリエータ |
| `UNLISTEN` | | リスナーを解除するアクション（指定しない場合はリッスンし続ける） |

以下に `UNLISTEN` が指定された場合の `runLoop` によるループ処理の流れを説明する。

- `UNLISTEN` の発生と Event Channel からのイベントの発生を同時に監視する。
  - チャンネルからのイベントが発生した場合， PHP側の `broadcastWith()` で指定されたペイロードがJSON変換経由で `payload` にセットされる。これをアクションクリエータのペイロードとしてそのまま渡し，アクションをディスパッチする。
  - `UNLISTEN` として指定されたイベントが発生した場合，ループを脱出して Event Channel を破棄する。

これらの一連のコードにより，PHP側からイベントがブロードキャストされたとき，指定された Redux のアクションをディスパッチすることができる。

#### （発展） Presence Channel への対応

アーキテクチャ的に噛み合わない部分があったので Presence Channel への対応は見送ろうと思ったが，[`SocketIoPresenceChannel`](https://github.com/laravel/echo/blob/f38bf48414d4553ba93cc6e976dfb8be1ad3f440/src/channel/socketio-presence-channel.ts) の実装を見ると， Private Channel に少し毛が生えた程度なので以下のように少し自分で書くだけで対応できそう。

複数のチャットルームに同時に入室できるみたいな実装だと，どのチャンネルを出たのかの判定が必要なので，その場合はもうちょっと `runLoop` 関数が複雑になるので注意。（ここでは割愛）

```js
export function *joinChatroom(roomId) {
  yield* forkForPrivateChannel(
    `chatrooms.${roomId}`,
    'presence:subscribed',
    (members) => initChatroomMembers(members.map(m => m.user_info)),
     LEAVE_CHATROOM,
  )
  yield* forkForPrivateChannel(
    `chatrooms.${roomId}`,
    'presence:joining',
    (member) => chatroomMemberJoining(member.user_info),
     LEAVE_CHATROOM,
  )
  yield* forkForPrivateChannel(
    `chatrooms.${roomId}`,
    'presence:leaving',
    (member) => chatroomMemberLeaving(member.user_info),
     LEAVE_CHATROOM,
  )
}
```

[`Echo`]: https://github.com/laravel/echo/blob/f38bf48414d4553ba93cc6e976dfb8be1ad3f440/src/echo.ts
[`WebSocket`]: https://developer.mozilla.org/ja/docs/Web/API/WebSockets_API
[`axios`]: https://github.com/axios/axios
[`redux-saga`]: https://github.com/redux-saga/redux-saga