gRPC / MagicOnion 入門 (17) - 切断検知と自動再接続

gRPC はサーバーとクライアントが常時コネクションを張っている状態です。このコネクションが切断されたタイミングを検知して後処理や再接続処理をしたい、というのはよくパターンかと思います。実は、生の gRPC で切断検知をするのは実はかなり面倒です。MagicOnion はそのあたりを上手くラップ*1し、扱いやすい形として提供してくれています。

今回は、それらを利用した切断検知と再接続の手法について見ていきます。

クライアント側で切断を検知

例えば、サーバーがダウンしたりなどしてコネクションが切断されたことをクライアント側で検知する方法は以下のようにします。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);
    var context = new ChannelContext(channel, () => "xin9le");

    //--- 切断検知を仕込む
    context.RegisterDisconnectedAction(() =>
    {
        Console.WriteLine("Disconnected");
    });

    //--- 接続待機
    await context.WaitConnectComplete();

    //--- 接続されてから 10 秒以内にサーバーを落とすと「Disconnected」と表示される
    await Task.Delay(10000);
}

ChannelContext.RegisterDisconnectedAction で切断のタイミングをフックすることができます！超簡単！

また、ここまで長らくお行儀悪く書いてこなかったのですが、ChannelContext は Dispose するのが良いです。以下の例のように Dispose しても切断検知が走ります。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);
    var context = new ChannelContext(channel, () => "xin9le");

    //--- 切断検知を仕込む
    context.RegisterDisconnectedAction(() =>
    {
        Console.WriteLine("Disconnected");
    });

    //--- 接続待機
    await context.WaitConnectComplete();

    //--- 切断する
    Console.WriteLine("1");
    await Task.Delay(1000);
    Console.WriteLine("2");

    context.Dispose();  // 切断！

    Console.WriteLine("3");
    await Task.Delay(1000);
    Console.WriteLine("4");

    Console.ReadLine();
}

//--- 結果
/*
1
2
3
Disconnected
4
*/

チャンネルのシャットダウン

これまたお行儀悪くずっと書いてこなかったのですが、gRPC の Channel はプロセスを終了する前にシャットダウンすることが強く推奨されています。また、シャットダウンを検知して後処理を行うこともできるようになっています。以下のような感じです。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);

    //--- シャットダウン検知
    channel.ShutdownToken.Register(() =>
    {
        Console.WriteLine("Shutdown");
    });

    //--- チャンネルをシャットダウン
    Console.WriteLine("1");
    await channel.ShutdownAsync();
    Console.WriteLine("2");

    Console.ReadLine();
}


//--- 結果
/*
1
Shutdown
2
*/

MagicOnion が提供する ChannelContext を利用している場合、終了処理は以下のような感じになると思います。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);
    var context = new ChannelContext(channel, () => "xin9le");

    //--- シャットダウン検知 / 切断検知
    channel.ShutdownToken.Register(() =>
    {
        Console.WriteLine("Shutdown");
    });
    context.RegisterDisconnectedAction(() =>
    {
        Console.WriteLine("Disconnected");
    });

    //--- 接続待機
    await context.WaitConnectComplete();
    await Task.Delay(1000);

    //--- 切断する
    Console.WriteLine("1");
    context.Dispose();
    Console.WriteLine("2");
    await channel.ShutdownAsync();
    Console.WriteLine("3");

    Console.ReadLine();
}

//--- 結果
/*
1
2
Shutdown
Disconnected
3
*/

サーバー側で切断検知

サーバー側でも接続していたクライアントがいなくなったことを検知して後処理を行いたいケースはよくあります。サーバー側での検知は以下のように行います。

public class SampleApi : ServiceBase<ISampleApi>, ISampleApi
{
    public async UnaryResult<Nil> Sample()
    {
        this.GetConnectionContext().ConnectionStatus.Register(() =>
        {
            Console.WriteLine("Disconnect detected!!");
        });
        return Nil.Default;
    }
}

ConnectionContext.ConnectionStatus は CancellationToken 型になっていて、クライアントの切断が検知されたときに Cancel が発行される仕組みになっています。その Cancel に反応できるように Register メソッドで事前に処理を登録しておく感じです。

例えばクライアント側を以下のように実装したとすると、ChannelContext.Dispose を呼び出したタイミングでサーバー側で切断検知され「Disconnected detected!!」が表示されます。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);
    var context = new ChannelContext(channel, () => "xin9le");

    await context.WaitConnectComplete();
    var client = context.CreateClient<ISampleApi>();
    await client.Sample();  // サーバー側で切断検知できるようにする
    await Task.Delay(1000);

    context.Dispose();  // ここを呼び出すとサーバー側で切断検知が走る
    await channel.ShutdownAsync();
    Console.ReadLine();
}

クライアントの自動再接続を行う

トンネルに入って出たときや、サービスの一時的なダウンから復旧した場合などは、自動的に再接続して復旧してほしいものです。そう言った処理も先の切断検知のタイミングを利用すれば実現できます。例えば以下のような感じです。

static async Task MainAsync()
{
    var channel = new Channel("localhost", 12345, ChannelCredentials.Insecure);
    var context = new ChannelContext(channel, () => "xin9le");

    //--- 再接続処理
    context.RegisterDisconnectedAction(async () =>
    {
        Console.WriteLine("Reconnecting...");
        await context.WaitConnectComplete();  // 再接続待ち
        Console.WriteLine("Reconnected");
    });

    //--- 接続待機
    Console.WriteLine("Connecting...");
    await context.WaitConnectComplete();
    Console.WriteLine("Connected");

    //--- この待ち時間にサーバーを落としたり立ち上げたりしてみましょう
    await Task.Delay(30000);

    //--- 切断する
    Console.WriteLine("Shutdown");
    context.Dispose();
    await channel.ShutdownAsync();
    Console.ReadLine();
}

//--- 実行例
/*
Connecting...
Connected
Reconnecting...
Reconnected
Shutdown
*/