TypeScriptでgRPCのstreaming RPCを使ったチャットのサンプル

はじめに

Node.jsのgRPCサンプルコードを解説されている方がいたのですが、この記事で触れられているのはUnaryなリクエストのみで、Streem通信に関しては触れられていなかったのでChartを例に試してみた。

Node.jsは現時点(2018/12/23)でのLTSv10.14.2を使っています。 gRPCの通信方式の違いに関しては、公式ドキュメントの RPC life cycleの項目が参考になります。

Protpcol Buffersの定義

gRPC公式サンプルの中に含まれているstreamを使った例 route_guide.proto を参考にchartアプリケーションを想定した簡単なProtpcol Buffersの定義を作成しました。 https://github.com/grpc/grpc/tree/master/examples/protos

// proto version.
syntax = "proto3";

// package name.
package example;

service Chat {
  rpc join(stream Message) returns (stream Message) {}
  rpc send(Message) returns (Message) {}
}

message Message {
  string user = 1;
  string text = 2;
}

上記.protoファイルのmessageと対応するTypeScriptのInterfaceも作成しました。

// .protoファイルの message定義に合せてtypescriptのinterfaceを用意
export interface Message {
  user: string;
  text: string;
}

Server

今回は、Node.jsの grpc パッケージのインターフェースを一通り眺めながら進めて行きたかったので、protoファイルからコードを自動生成を行うことはしていません。protoLoaderに.protoファイルを渡してサービス定義を読み込みます。
今回のように実行時に動的にコード生成を行う方法を dynamic codegen といい、事前にコード生成(静的にコード生成)をおこなう方法を static codegen と呼ぶようです。

import * as grpc from 'grpc';
import { ServerWriteableStream } from 'grpc';
import * as protoLoader from '@grpc/proto-loader';

import { Message } from './types';

const server = new grpc.Server();
const serverAddress = '0.0.0.0:5001';

// .protoファイルを動的に読み込み
const proto = grpc.loadPackageDefinition(
  protoLoader.loadSync('./chat.proto', {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true,
  })
);

const users: ServerWriteableStream<Message>[] = [];

function join(call: ServerWriteableStream<Message>): void {
  users.push(call);
  notifyChat({ user: 'Server', text: `new user joined ...` });
}

function send(call: ServerWriteableStream<Message>): void {
  console.log(`send message from ${call.request.user}: ${call.request.text}`);
  notifyChat(call.request);
}

function notifyChat(message: Message): void {
  // すべてのユーザーにメッセージを送信
  users.forEach((user) => {
    user.write(message);
  });
}


// .protoファイルで定義したserviceと上記実装をマッピング
server.addService(proto.example.Chat.service, {
  join: join,
  send: send
});

// Serverのconfig設定 & 起動
server.bind(serverAddress, grpc.ServerCredentials.createInsecure());
server.start();

Client

Client側もServer側と同様に、protoLoaderに.protoファイルを渡してサービス定義を読み込みます。

import * as grpc from 'grpc';
import * as protoLoader from '@grpc/proto-loader';
import * as readline from 'readline';

import { Message } from './types';

const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout,
});

const proto = grpc.loadPackageDefinition(
  protoLoader.loadSync('./chat.proto', {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true,
  })
);

const remoteServer = '0.0.0.0:5001';

let username = '';

const client = new proto.example.Chat(
  remoteServer,
  grpc.credentials.createInsecure()
)

function startChat(): void {
  let channel = client.join({ user: username });

  channel.on('data', onDataCallback);

  // 標準入力に新しい行が入力されるごとにServerに送信
  rl.addListener('line', (text: string) => {
    client.send({ user: username, text: text }, () => {});
  });
}


function onDataCallback(message: Message): void {
  if (message.user === username) {
    return;
  }

  console.log(`${message.user}: ${message.text}`);
}

rl.question('User name: ', (answer: string) => {
  username = answer;

  startChat();
});

動作確認

今回は、REPLで実行する。

まずはServer側を実行しておく

$ ts-node server.ts

その後、Client側(1)を実行
ユーザー名を入力しEnterを押すと入力待受状態になる

$ ts-node client.ts
User name: daikiojm
Server: new user joined ...

Client(2)も同様に実行

$ ts-node client.ts
User name: test-user
Server: new user joined ...

その後は双方向の通信が確立される

$ ts-node client.ts
User name: test-user
Server: new user joined ...
daikiojm: はらへった
me too

終わりに

gRPCのstreaming RPCはClient - Serverで真意を発揮しそうなので、機会があればgRPC-webでも試してみたい。
従来httpとwsを併用していたような箇所をgRPC1本にまとめつつ型安全を保てるのはかなり実用的な印象がある。
Node.jsでのServer実装の例は少なそうなので、そのうちもう少し大きめのサンプルを作ってみたいかもしれない。