RubyでgRPCのストリーミング

概要

• gRPCの双方向なストリーミング機能をRubyで利用する方法

前提

• 以下でgRPC用のgemをインストール

$ gem install grpc
$ gem install grpc-tools

方法

protoファイル

• service定義で、パラメータと戻り値に"stream"を付ける

sample.proto

syntax = 'proto3';

package sample;

service Echo {
    rpc Talk (stream Request) returns (stream Response);
}

message Request {
    string message = 1;
}

message Response {
    string message = 1;
}

コンパイル

grpc_tools_ruby_protoc --ruby_out=lib --grpc_out=lib sample.proto

• 上記protoファイルをコンパイル
  • sample_pb.rbとsample_services_pb.rbが生成される

プログラム実装（例）

• サーバ側、クライアント側双方、Enumeratorを介してやりとりする

サーバ側

sample_server.rb

#!/usr/bin/env ruby

this_dir = File.expand_path(File.dirname(__FILE__))
lib_dir = File.join(this_dir, 'lib')
$LOAD_PATH.unshift(lib_dir) unless $LOAD_PATH.include?(lib_dir)

require 'grpc'
require 'sample_services_pb'

class EchoServer < Sample::Echo::Service
  def talk(requests)
    return enum_for(:talk, requests) unless block_given?

    requests.each do |request|
      sleep rand(5)
      message = "received (#{request.message})"
      yield Sample::Response.new(message: message)
    end
  end
end

def main
  s = GRPC::RpcServer.new
  s.add_http2_port('localhost:50051', :this_port_is_insecure)
  s.handle(EchoServer)
  s.run_till_terminated
end

main

• サーバ用クラスのメソッドの引数には、クライアント側から渡されたEnumeratorなオブジェクト（後述）が渡ってくる
• 上記Enumeratorオブジェクトに対して、eachでブロックを渡す
  • yieldでレスポンス用のオブジェクトを渡すようにしておく
• メソッドの戻り値としてEnumeratorオブジェクトを返す
  • 実行時にはブロックなしで呼ばれるので、ブロックなしで呼ばれたらObject#enum_forで自メソッドをEnumerator化して返すようにしておく
  • このEnumeratorオブジェクトがクライアント側のスタブの戻り値になる

クライアント側

• サーバ側がObject#enum_forだったので、サンプル的にEnumerator.newで

sample_client.rb

#!/usr/bin/env ruby

this_dir = File.expand_path(File.dirname(__FILE__))
lib_dir = File.join(this_dir, 'lib')
$LOAD_PATH.unshift(lib_dir) unless $LOAD_PATH.include?(lib_dir)

require 'grpc'
require 'sample_services_pb'

def main
  requests = Enumerator.new do |yielder|
    loop do
      print "\ninput: "
      message = STDIN.gets.chomp
      next if message.size == 0
      break if %w(quit exit).include?(message)

      yielder << Sample::Request.new(message: message)
    end
  end

  stub = Sample::Echo::Stub.new('localhost:50051', :this_channel_is_insecure)

  responses = stub.talk(requests)
  responses.each do |response|
    puts "\nresponse: #{response.message}"
  end
end

main

• リクエスト用にスタブのメソッドにEnumeratorオブジェクトを渡す
  • これがサーバ側メソッドの引数になる
• スタブの戻り値にサーバ側から返したEnumeratorオブジェクトが渡ってくる
  • each実行でサーバ側メソッド（のenum_forでEnumerator化されたもの）が実行される
  • サーバ側メソッド内でeachが実行されることで、リクエスト用に渡したクライアント側のEnumerator(Enumerator.newしたもの)が実行される
  • Enumerator.newのブロック内でEnumerator::Yielder#<<に渡したリクエスト用オブジェクトがサーバ側のeachのブロック引数として渡る
  • サーバ側でyieldに渡されたレスポンス用オブジェクトが、クライアント側のeachのブロック引数として渡ってくる