【Unity】MessagePipeを触ってみる2 ~さまざまなIPublisher/ISubscriber編~ 

この記事は続きです。前回はこちら。

[【Unity】MessagePipeを触ってみる1 ~概要・導入編~](https://qiita.com/toRisouP/items/4775f15a6612ab765892)


この記事では`VContainer`の使用を前提としています。

記事に登場するコード類はこちらのリポジトリで公開しています。

* https://github.com/TORISOUP/MessagePipe_Samples

---

# さまざまなIPublisher/ISubscriber

`MessagePipe`では`IPublisher<T>`/`ISubscriber<T>`を用いてメッセージの送信と受信を行います（長いので`IPub/ISub`と略します）。

この`IPub/ISub`ですがいくつかバリエーションが用意されており、それぞれで挙動が異なります。
今回はそのバリエーションを紹介します。

## `IPublisher<T>`/`ISubscriber<T>`

### 概要

まずはもっともベーシックなものです。
指定した型`T`のデータを同期でそのまま即時伝達を行います。

![Normal.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/263eec12-988b-f379-325b-4bac0f956204.jpeg)


### 例

```cs:送信用データ
using System;

namespace MessagePipeSample.PubSub.Normal
{
    /// <summary>
    /// 送信データ（適当に定義した型）
    /// </summary>
    public readonly struct SendData : IEquatable<SendData>
    {
        public int Id { get; }
        public int Value { get; }

        public SendData(int id, int value)
        {
            Id = id;
            Value = value;
        }
        
        public bool Equals(SendData other)
        {
            return Id == other.Id && Value == other.Value;
        }

        public override bool Equals(object obj)
        {
            return obj is SendData other && Equals(other);
        }

        public override int GetHashCode()
        {
            return HashCode.Combine(Id, Value);
        }
    }
}
```

```cs:送受信を行うクラス
using System;
using MessagePipe;
using UnityEngine;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Normal
{
    public class ReadWriteSample1 : IInitializable, IDisposable, ITickable
    {
        private readonly IPublisher<SendData> _publisher;
        private readonly ISubscriber<SendData> _subscriber;
        private IDisposable _disposable;

        // それぞれをDIしてもらう
        public ReadWriteSample1(
            IPublisher<SendData> publisher,
            ISubscriber<SendData> subscriber)
        {
            _publisher = publisher;
            _subscriber = subscriber;
        }

        public void Initialize()
        {
            // 同じクラス内でPub/Subするのはナンセンスだけど、
            // 使い方のサンプルということでゆるして

            // SendDataを購読し、受信したらログにだす
            _disposable = _subscriber.Subscribe(data =>
            {
                // 受信データのうち、Idが一致するもののみ処理する
                if (data.Id == 0)
                {
                    Debug.Log($"Received: Id={data.Id} Value={data.Value}");
                }
            });
        }

        public void Tick()
        {
            // 毎フレーム送信する
            _publisher.Publish(new SendData(id: 0, UnityEngine.Random.Range(0, 100)));
        }

        public void Dispose()
        {
            // 購読中止
            _disposable?.Dispose();
        }
    }
}
```

```cs:DI
using MessagePipe;
using VContainer;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Normal
{
    public class GameLifetimeScope : LifetimeScope
    {

        protected override void Configure(IContainerBuilder builder)
        {
            // MessagePipeの設定
            var options = builder.RegisterMessagePipe();

            // SendDataを伝達できるように設定する
            builder.RegisterMessageBroker<SendData>(options);

            // ReadWriteSample1を起動
            builder.RegisterEntryPoint<ReadWriteSample1>(Lifetime.Singleton);
            
        }
    }
}
```


#### 実行結果

毎フレーム、ログが繰り返し表示され続けます。

```
Received: Id=0 Value=2
Received: Id=0 Value=59
Received: Id=0 Value=21
Received: Id=0 Value=86
...(以下実行を停止するまで続く)
```

`IPublisher<T>`/`ISubscriber<T>`はすごくシンプルに、`Publish()`されたときのデータを即時・同期ですべての`Subscriber`へと伝達します。これにより「非常にシンプルな形でのメッセージのグローバル配信」が可能となります。

### 考察：型が同じだが用途が異なる場合に、`Pub/Sub`を別けて扱えるのか

`IPublisher<T>`/`ISubscriber<T>`は型を指定して送受信はできるが、これを「グループ分けして使えるのか」という疑問です。
（「グループAにはそれ専用の`IPub/ISub`」を、「グループBにはそれ専用のそれを」といった区別ができるのか）

#### 結論：`IPublisher<TKey, T>`/`ISubscriber<TKey, T>`を使おう

`IPublisher<TKey, T>`/`ISubscriber<TKey, T>`という、`Key`を指定して送受信できるものが用意されているのでこちらを使いましょう。

使い方については後述します。

#### 別解A: データの中に識別子を入れる

何らかの理由で`IPublisher<TKey, T>`/`ISubscriber<TKey, T>`が使えない場合にどうするか。簡単なやり方としては、送信するデータの中に識別用の何かを入れて、受信側でハンドリングするというやり方です。

```cs:受け取ってからフィルタリング
_disposable = _subscriber.Subscribe(data =>
{
    // 受信データのうち、Idが自分と一致するもののみ処理する
    if (data.Id == _myId)
    {
        Debug.Log($"Received: Id={data.Id} Value={data.Value}");
    }
});
```

この場合は`Subscriber`側ですべてのデータを受けて、そこから不要なデータは捨てるという挙動になります。そのため`Subscriber`の数が膨大な場合は無駄な負荷が発生します。


#### 別解B: 用途ごとに型を変える

前提からひっくり返すパターン。しかし設計上、「別の意味・用途をもつデータ構造なら概念ごと別ける」というは重要だったりします。なので場合によってはこれが最適解のパターンになることもありそう。


```cs:型をそもそも別ける
// 用途が違うならそもそも型から別けてしまうべき
builder.RegisterMessageBroker<SendData1>(options);
builder.RegisterMessageBroker<SendData2>(options);
```

## `IPublisher<TKey, T>`/`ISubscriber<TKey, T>`

### 概要

挙動は`IPublisher<T>`/`ISubscriber<T>`とだいたい同じなのですが、`TKey`を指定して送受信のグルーピングが可能になっています。

`Publish`時に指定した`TKey`と同じ値で`Subscribe`している対象にのみメッセージが届くように、`MessageBroker`の内部でフィルタリングが実行されます。このため、関係ない`Subscriber`に不必要にデータを送ることがなくなり負荷軽減にも繋がります。

![Key.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/3c57dbe3-21bf-a1d2-3709-725dfec13104.jpeg)
（同じインタフェースを用いて`Pub/Sub`するが…）

![Key2.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/c4d639ee-5cca-2be5-db41-3af07972be0f.jpeg)
（`MessageBroker`の内部でグループ分けされるため、異なる`TKey`の`Subscriber`にはメッセージ送信はされない）


なお内部実装的には`TKey`をKeyとした`Dictionary`が用いられています。そのため`TKey`として扱うオブジェクトには適切に`GetHashCode`を実装することを忘れないでください。

### 例

```cs:Key
using System;

namespace MessagePipeSample.PubSub.Key
{
    /// <summary>
    /// 識別子
    /// </summary>
    public readonly struct KeyData : IEquatable<KeyData>
    {
        public int _raw { get; }

        public KeyData(int raw)
        {
            _raw = raw;
        }

        public bool Equals(KeyData other)
        {
            return _raw == other._raw;
        }

        public override bool Equals(object obj)
        {
            return obj is KeyData other && Equals(other);
        }

        public override int GetHashCode()
        {
            return _raw;
        }
    }
}
```

```cs:メッセージ本体
using System;

namespace MessagePipeSample.PubSub.Key
{
    /// <summary>
    /// 送信データ（適当に定義した型）
    /// </summary>
    public readonly struct SendData : IEquatable<SendData>
    {
        public string Value { get; }

        public SendData(string value)
        {
            Value = value;
        }

        public bool Equals(SendData other)
        {
            return Value == other.Value;
        }

        public override bool Equals(object obj)
        {
            return obj is SendData other && Equals(other);
        }

        public override int GetHashCode()
        {
            return (Value != null ? Value.GetHashCode() : 0);
        }
    }
}
```

```cs:送信側
using MessagePipe;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Key
{
    /// <summary>
    /// 順繰りにデータを送る
    /// </summary>
    public sealed class RoundRobinSender : ITickable
    {
        private readonly IPublisher<KeyData, SendData> _publisher;

        private int _roundRobinIndex;

        public RoundRobinSender(IPublisher<KeyData, SendData> publisher)
        {
            _publisher = publisher;
        }

        public void Tick()
        {
            // 0,1,2 を順繰りにデータ送信する
            _roundRobinIndex = (_roundRobinIndex + 1) % 3;

            var key = new KeyData(_roundRobinIndex);
            var sendData = new SendData($"Key={_roundRobinIndex},Value={UnityEngine.Random.Range(0, 100)}");

            // Keyを指定して送信
            _publisher.Publish(key, sendData);
        }
    }
}
```

```cs:受信側
using System;
using MessagePipe;
using UnityEngine;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Key
{
    public sealed class Receiver : IInitializable, IDisposable
    {
        private readonly ISubscriber<KeyData, SendData> _subscriber;
        private IDisposable _disposable;

        public Receiver(ISubscriber<KeyData, SendData> subscriber)
        {
            _subscriber = subscriber;
        }

        public void Initialize()
        {
            // まとめて破棄するやつ
            var bag = DisposableBag.CreateBuilder();

            // --- Key = 0 のみを購読 ---
            _subscriber.Subscribe(new KeyData(0), data =>
                {
                    Debug.Log($"MyKey=0, ReceivedData=[{data.Value}]");
                })
            .AddTo(bag);

            // --- Key = 1 のみを購読 ---
            _subscriber.Subscribe(new KeyData(1), data =>
                {
                    Debug.Log($"MyKey=1, ReceivedData=[{data.Value}]");
                })
            .AddTo(bag);

            // --- Key = 2 は購読しない ---

            _disposable = bag.Build();
        }

        public void Dispose()
        {
            _disposable?.Dispose();
        }
    }
}
```

```cs:DI
using MessagePipe;
using VContainer;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Key
{
    public class GameLifetimeScope : LifetimeScope
    {
        protected override void Configure(IContainerBuilder builder)
        {
            // MessagePipeの設定
            var options = builder.RegisterMessagePipe();
            
            // KeyDataでグループ別けしてSendDataを送受信する
            builder.RegisterMessageBroker<KeyData, SendData>(options);

            builder.RegisterEntryPoint<RoundRobinSender>();
            builder.RegisterEntryPoint<Receiver>();
        }
    }
}
```

#### 実行結果

`KeyData`を用いてフィルタリングされた形でメッセージの受信が実行されます。

```
MyKey=1, ReceivedData=[Key=1,Value=92]
MyKey=0, ReceivedData=[Key=0,Value=28]
MyKey=1, ReceivedData=[Key=1,Value=13]
MyKey=0, ReceivedData=[Key=0,Value=78]
MyKey=1, ReceivedData=[Key=1,Value=2]
MyKey=0, ReceivedData=[Key=0,Value=44]
...(以下、Key=0/1を交互に繰り返し)
```

### 考察: Subscriberが居ないTKeyへのPublishはどうなるのか

「誰も購読していない`TKey`」を指定して`Publish`を行うとどうなるのでしょうか。

#### 結論: 送信すらされない

`IPublisher<TKey, T>`/`ISubscriber<TKey, T>`では、`TKey`に応じて別けて`Subscriber`を管理しています。そして`Subscriber`が存在しない`TKey`については、そもそも送信処理が実行されず無視されるようになっています。

* [MessageBroker<TKey, T>の実装の中身](https://github.com/Cysharp/MessagePipe/blob/master/src/MessagePipe/MessageBroker_Key.cs#L56-L59)



## `IBufferedPublisher<T>`/`IBufferedSubscriber<T>`

### 概要

`IBufferedPublisher<T>`/`IBufferedSubscriber<T>`は「直前に`Publish`された値を1つだけ保持する」という挙動をします。つまり「`Subscribe`時に最新の値があればそれを発行し、以降は通常の`IPub`/`ISub`と同じ」という動作をします。

なお`T`が`class`であり、かつ`null`が`Publish`されている（または1度も`Publishしていない`）場合は`Subscribe`時の最新値の発行は行われません。

### 例

```cs:送信データ
using System;

namespace MessagePipeSample.PubSub.Buffered
{
    /// <summary>
    /// 送信データ
    /// </summary>
    public readonly struct SendData : IEquatable<SendData>
    {
        public string Value { get; }

        public SendData(string value)
        {
            Value = value;
        }

        public bool Equals(SendData other)
        {
            return Value == other.Value;
        }

        public override bool Equals(object obj)
        {
            return obj is SendData other && Equals(other);
        }

        public override int GetHashCode()
        {
            return (Value != null ? Value.GetHashCode() : 0);
        }
    }
}
```

```cs:送信側
using System;
using Cysharp.Threading.Tasks;
using MessagePipe;
using UnityEngine;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Buffered
{
    public sealed class BufferSender : IInitializable
    {
        private readonly IBufferedPublisher<SendData> _bufferedPublisher;

        public BufferSender(IBufferedPublisher<SendData> bufferedPublisher)
        {
            _bufferedPublisher = bufferedPublisher;
        }

        public void Initialize()
        {
            // 2回送信する
            _bufferedPublisher.Publish(new SendData("1"));
            // この2度目がBufferに積まれた状態でしばらく置かれる
            _bufferedPublisher.Publish(new SendData("2"));
            
            Debug.Log("Sender: Message sent.");
            
            UniTask.Void(async () =>
            {
                Debug.Log("Wait for a second...");
                await UniTask.Delay(TimeSpan.FromSeconds(1));

                // しばらくたったから次の値をPublish
                _bufferedPublisher.Publish(new SendData("3"));
            });
        }
    }
}
```

```cs:受信側
using System;
using MessagePipe;
using UnityEngine;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Buffered
{
    public sealed class BufferReceiver : IStartable, IDisposable
    {
        private readonly IBufferedSubscriber<SendData> _bufferedSubscriber;
        private IDisposable _disposable;

        public BufferReceiver(IBufferedSubscriber<SendData> bufferedSubscriber)
        {
            _bufferedSubscriber = bufferedSubscriber;
        }

        // Sender.Initialize()よりこっちの方が実行開始が遅い
        public void Start()
        {
            Debug.Log("BufferReceiver: Subscribe start.");
            _disposable = _bufferedSubscriber.Subscribe(data =>
            {
                Debug.Log($"Received! :{data.Value}");
            });
        }

        public void Dispose()
        {
            _disposable?.Dispose();
        }
    }
}
```

```cs:DI
using MessagePipe;
using VContainer;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Buffered
{
    public class GameLifetimeScope : LifetimeScope
    {
        protected override void Configure(IContainerBuilder builder)
        {
            // MessagePipeの設定
            var options = builder.RegisterMessagePipe();
            
            // KeyDataでグループ別けしてSendDataを送受信する
            builder.RegisterMessageBroker<SendData>(options);

            builder.RegisterEntryPoint<BufferSender>();
            builder.RegisterEntryPoint<BufferReceiver>();
        }
    }
}
```

#### 実行結果

```
// この前に発行された「1」は無視される
Sender: Message sent.    // この時点で「2」がBufferに積まれている
Wait for a second... 
BufferReceiver: Subscribe start.
Received! :2            // Bufferされていた「2」が出力
Received! :3            // 後にPublishされた「3」
```

## `IAsyncPublisher<T>`/`IAsyncSubscriber<T>`

### 概要

`IAsyncPublisher<T>`/`IAsyncSubscriber<T>`（長いので`IAP<T>`/`IAS<T>`と書きます）は、`IPub<T>`/`ISub<T>`の非同期版です。何が非同期なのかというと、「**メッセージの送信時に、受信側の処理を完了するのを待つ**」の部分が非同期になっています。

### AsyncPublishStrategy

`IAP<T>`/`IAS<T>`は`AsyncPublishStrategy`を指定することで非同期処理の戦略を変更することができます。

#### `AsyncPublishStrategy.Parallel`

`Parallel`を指定した場合は並行動作となります。

`PublishAsync`したタイミングで各`Subscriber`へ一斉にメッセージを発送し、`Subscriber`側の非同期処理がすべて終わるのを待機します。
（一斉に非同期処理を実行して`UniTask.WhenAll`で待ち受けるイメージ）


![Parallel.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/bc983f53-15e2-12e2-25cc-56cf8465b550.jpeg)


#### `AsyncPublishStrategy.Sequential`

`Sequential`を指定した場合は直列動作となります。

`PublishAsync`したタイミングで各`Subscriber`へ1つずつ順番にメッセージを発送します。発送先の`Subscriber`で非同期処理が終わるのを待機し、完了したら次の`Subscriber`へ発送するという処理を繰り返します。

![Sequential.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/fd33132d-30b8-02bf-b52f-dfd960310b13.jpeg)



### 実装例

実際の動きを見たほうがわかりやすいと思うので実装例を。

* シーン上にCubeが配置されている
* 「リセットボタン」でCubeの位置がランダムに初期化される
* 「Sequential」を押すと**AsyncPublishStrategy.Sequential**でCubeをゴールに移動させるメッセージを送る
* 「Parallel」を押すと**AsyncPublishStrategy.Parallel**でCubeをゴールに移動させるメッセージを送る
* **移動中は各ボタン操作を無効化する**

という挙動を`IAP<T>`/`ISP<T>`を用いて実装してみます。

![move.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/31c82241-ef07-bf07-0934-916a94367563.gif)



#### 送信メッセージ

今回は「ゴール先の座標送信」と「位置のリセット」で2つメッセージを用意します。

```cs:送信データ（ゴール先情報）
using System;
using UnityEngine;

namespace MessagePipeSample.PubSub.Async
{
    /// <summary>
    /// 目的地情報
    /// </summary>
    public readonly struct TargetPosition : IEquatable<TargetPosition>
    {
        public Vector3 Position { get; }

        public TargetPosition(Vector3 position)
        {
            Position = position;
        }

        public bool Equals(TargetPosition other)
        {
            return Position.Equals(other.Position);
        }

        public override bool Equals(object obj)
        {
            return obj is TargetPosition other && Equals(other);
        }

        public override int GetHashCode()
        {
            return Position.GetHashCode();
        }
    }
}
```

```cs:送信データ（リセットイベント）
namespace MessagePipeSample.PubSub.Async
{
    // 状態リセットイベント
    public readonly struct ResetEvent
    {
        public static readonly ResetEvent Default = new ResetEvent();
    }
}
```

#### 移動するCube

`Cube`側に貼り付けるコンポーネントです。
`IAsyncSubscriber<TargetPosition>`からゴール先の座標を受け取ったら、**非同期処理として移動処理を実行します。**

```cs:移動するCubeのコンポーネント
using System.Threading;
using Cysharp.Threading.Tasks;
using MessagePipe;
using UnityEngine;
using VContainer;

namespace MessagePipeSample.PubSub.Async
{
    public sealed class MovingCube : MonoBehaviour
    {
        [Inject] private IAsyncSubscriber<TargetPosition> _asyncTargetSubscriber;
        [Inject] private ISubscriber<ResetEvent> _resetEventSubscriber;

        private readonly float _speed = 5.0f;

        private void Start()
        {
            _asyncTargetSubscriber.Subscribe(async (target, ct) =>
                {
                    await MoveToTargetAsync(target.Position, ct);
                })
                .AddTo(this.GetCancellationTokenOnDestroy());

            _resetEventSubscriber.Subscribe(_ => ResetPosition())
                .AddTo(this.GetCancellationTokenOnDestroy());
        }

        /// <summary>
        /// 指定座標に向かってゆっくり移動する
        /// </summary>
        private async UniTask MoveToTargetAsync(Vector3 target, CancellationToken ct)
        {
            while (!ct.IsCancellationRequested)
            {
                // 目標地点との差分
                var delta = (target - transform.position);

                // 1.0m以内なら到着
                if (delta.sqrMagnitude < 1.0f)
                {
                    return;
                }

                // 離れているなら目標に向かって移動する
                transform.position += _speed * (delta.normalized) * Time.deltaTime;

                await UniTask.Yield();
            }
        }

        private void ResetPosition()
        {
            transform.position =
                new Vector3(Random.Range(-5f, 5f), 0, Random.Range(-5f, 5f));
        }
    }
}
```

#### UI操作部分

`uGUI`の`Button`と連動してメッセージを発行します。
（[GetAsyncClickEventHandlerの使い方はこちらを参考に](https://neue.cc/2018/12/25_571.html)）

ここで「移動ボタン」が押されたときに**各Cubeへメッセージを送り、それぞれの移動が終わるのを待つ**という処理を`IAP<T>/IAS<T>`で実現しています。

```cs:uGUIと連動してメッセージ送信
using System.Threading;
using Cysharp.Threading.Tasks;
using MessagePipe;
using UnityEngine;
using UnityEngine.UI;
using VContainer;
using Random = UnityEngine.Random;

namespace MessagePipeSample.PubSub.Async
{
    /// <summary>
    /// uGUIに連動してCubeに命令を出す
    /// </summary>
    public sealed class CubeController : MonoBehaviour
    {
        [Inject] private IAsyncPublisher<TargetPosition> _asyncPublisher;
        [Inject] private IPublisher<ResetEvent> _resetPublisher;

        // 各種ボタン
        [SerializeField] private Button _moveParallelButton;
        [SerializeField] private Button _moveSequentialButton;
        [SerializeField] private Button _resetButton;

        [SerializeField] private Transform _goalMarkerObject;

        private void Start()
        {
            var ct = this.GetCancellationTokenOnDestroy();
            
            // ボタンイベントの購読開始
            WaitForResetButtonAsync(ct).Forget();
            WaitForMoveParallelButtonAsync(ct).Forget();
            WaitForMoveSequentialButtonAsync(ct).Forget();
        }

        /// <summary>
        /// リセットボタンの処理
        /// </summary>
        private async UniTaskVoid WaitForResetButtonAsync(CancellationToken ct)
        {
            var asyncHandler = _resetButton.GetAsyncClickEventHandler(ct);

            while (!ct.IsCancellationRequested)
            {
                // ボタンがクリックされたらリセットメッセージを発行
                await asyncHandler.OnClickAsync();
                _resetPublisher.Publish(ResetEvent.Default);
            }
        }

        /// <summary>
        /// 移動ボタン（並行処理）
        /// </summary>
        private async UniTaskVoid WaitForMoveParallelButtonAsync(CancellationToken ct)
        {
            var asyncHandler = _moveParallelButton.GetAsyncClickEventHandler(ct);

            while (!ct.IsCancellationRequested)
            {
                await asyncHandler.OnClickAsync();

                // PublishAsyncが完了するまでボタンを無効化
                SwitchButtonInteractable(false);
                
                // ランダムに移動先を決定
                var target = new UnityEngine.Vector3(Random.Range(-5f, 5f), 0, Random.Range(-5f, 5f));
                _goalMarkerObject.transform.position = target;

                // Cubeにメッセージを送信し、移動が終わるまで待つ
                // AsyncPublishStrategy.Parallel を指定しているので並行実行
                await _asyncPublisher.PublishAsync(new TargetPosition(target),
                    AsyncPublishStrategy.Parallel, ct);

                // ボタンを再有効化
                SwitchButtonInteractable(true);
            }
        }

        /// <summary>
        /// 移動ボタン（直列処理）
        /// </summary>
        private async UniTaskVoid WaitForMoveSequentialButtonAsync(CancellationToken ct)
        {
            var asyncHandler = _moveSequentialButton.GetAsyncClickEventHandler(ct);

            while (!ct.IsCancellationRequested)
            {
                await asyncHandler.OnClickAsync();

                // PublishAsyncが完了するまでボタンを無効化
                SwitchButtonInteractable(false);

                // ランダムに移動先を決定
                var target = new UnityEngine.Vector3(Random.Range(-5f, 5f), 0, Random.Range(-5f, 5f));
                _goalMarkerObject.transform.position = target;

                // Cubeにメッセージを送信し、移動が終わるまで待つ
                // AsyncPublishStrategy.Sequential を指定しているので直列実行
                await _asyncPublisher.PublishAsync(new TargetPosition(target),
                    AsyncPublishStrategy.Sequential, ct);
                
                // ボタンを再有効化
                SwitchButtonInteractable(true);
            }
        }

        private void SwitchButtonInteractable(bool isEnabled)
        {
            _moveParallelButton.interactable = isEnabled;
            _moveSequentialButton.interactable = isEnabled;
            _resetButton.interactable = isEnabled;
        }
    }
}
```


#### DI設定

```cs:DI
using MessagePipe;
using UnityEngine;
using VContainer;
using VContainer.Unity;

namespace MessagePipeSample.PubSub.Async
{
    public class GameLifetimeScope : LifetimeScope
    {
        [SerializeField] private MovingCube _cubePrefab;

        protected override void Configure(IContainerBuilder builder)
        {
            // MessagePipeの設定
            var options = builder.RegisterMessagePipe();

            // デフォルトの非同期処理のやり方
            // PublishAsync時に個別指定もできる
            options.DefaultAsyncPublishStrategy = AsyncPublishStrategy.Parallel;

            builder.RegisterMessageBroker<TargetPosition>(options);
            builder.RegisterMessageBroker<ResetEvent>(options);

            // DIしながらInstantiate
            builder.RegisterBuildCallback(resolver =>
            {
                // 3つ生成する
                for (int i = 0; i < 3; i++)
                {
                    resolver.Instantiate(_cubePrefab);
                }
            });
        }
    }
}
```

#### 実行結果

[こちらで実際に動かせます](https://torisoup.github.io/MessagePipe_Samples/AsyncSample/index.html)

* 並行実行時

`AsyncPublishStrategy.Parallel`でメッセージ送信した場合の挙動です。
すべての`Cube`へ一斉にメッセージを渡して移動処理を行い、すべてが終わるまで待ちます。
![Parallel.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/17bc28c1-ec99-c84d-d952-b9333709036a.gif)



* 直列実行時

`AsyncPublishStrategy.Sequential`でメッセージ送信した場合の挙動です。
1つずつ順番に`Cube`へとメッセージを渡して移動処理を行い、それが終わったら次の`Cube`へ…という処理を終わるまで待ちます。


![sequential.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/44ffc5ef-91ff-86a6-8da6-1b90c655e0c2.gif)




### 考察

`IAP<T>`/`ISP<T>`は`Observable`では表現できなかった機能なので、`MessagePipe`の目玉機能だと思います。（一応、UniRxでも`AsyncMessageBroker`とかあったけど）

## `IAsyncPublisher<TKey, T>`/`IAsyncSubscriber<TKey, T>`

`IAP<T>`/`ISP<T>`の`Tkey`が指定できる版です。`IPublisher<TKey, T>`/`ISubscriber<TKey, T>`が非同期になったのと同等なので説明は省略します。


## `IBufferedAsyncPublisher<T>`/`IBufferedAsyncSubscriber<T>`

`IBufferedPublisher<T>`/`IBufferedSubscriber<T>`の非同期版です。

挙動は`IBufferedPublisher<T>`/`IBufferedSubscriber<T>`とだいたい同じですが、`Subscribe`自体が`Async`になっている点が異なります。

```cs

// SubscribeAsync()の戻り値はUniTask<IDisposable>
// Subscribe時に最新値があれば即発行されて非同期処理が実行される
var disposable = await _asyncBufferedTargetSubscriber.SubscribeAsync(
    async (target, ct) =>
    {
        await MoveToTargetAsync(target.Position, ct);
    });
```

## `IDistributedPublisher<TKey, TMessage>`/`IDistributedSubscriber<TKey, TMessage>`

`IDistributedPublisher<TKey, TMessage>`/`IDistributedSubscriber<TKey, TMessage>`は分散化された（ネットワークを跨いだ）メッセージ伝達に持ちいることができます。

長くなるのでこれの解説は別記事で。

# まとめ

どのパターンの`Pub/Sub`も便利なので、ユースケースに応じて適切なものを選択できるとよいでしょう。
個人的には`AsyncPub/AsyncSub`の挙動がかなり面白くて気に入っているので、使える場面があればガンガン使って行きたいと思います。