grpc_client.rb

require 'ds9'
require 'socket'
require 'uri'
require 'google/protobuf'
require 'io/wait'
require 'helloworld_pb'

class MySession < DS9::Client
  def initialize sock
    @sock      = sock
    @in_flight = {}
    @blocks    = {}
    super()
  end

  def on_stream_close id, errcode
    @blocks.delete(id).call @in_flight.delete(id)
  end

  def on_begin_headers frame
    if frame.headers?
      # We can have multiple header blocks, so lets just do this once
      @in_flight[frame.stream_id] ||= ''
    end
  end

  def on_data_chunk_recv id, data, flags
    @in_flight[id] << data
  end

  def send_event string
    @sock.write_nonblock string
  end

  def recv_event length
    case data = @sock.read_nonblock(length, nil, exception: false)
    when :wait_readable then DS9::ERR_WOULDBLOCK
    when nil            then DS9::ERR_EOF
    else
      data
    end
  end

  def run(&block)
    while want_read? || want_write?
      @sock.wait_readable
      receive
      send
    end
  end

  def submit_request headers, body, &block
    id = super
    @blocks[id] = block
    id
  end
end

# This URI comes from the generated rpc file
uri = URI.parse 'http://localhost:50051/helloworld.Greeter/SayHello'

socket = TCPSocket.new uri.host, uri.port
socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)

session = MySession.new socket
session.submit_settings []

def make_request
  # These classes come from the generated protobuf file
  req = Helloworld::HelloRequest.new(name: "world")
  data = Helloworld::HelloRequest.encode req
  x = [0, data.length, data].pack('CNa*')
end

def submit_request session, uri, &block
  path = uri.path == '' ? '/' : uri.path

  session.submit_request([
    [':scheme', uri.scheme],
    [':method', 'POST'],
    [':authority', [uri.host, uri.port].join(':')],
    [':path', path],
    ['te', 'trailers'],
    ['content-type', 'application/grpc'],
    ['user-agent', 'test'],
    ['grpc-accept-encoding', 'identity,deflate,gzip'],
    ['accept-encoding', 'identity,gzip'],
  ], make_request, &block)
end

5.times do
  submit_request(session, uri) do |res|
    compression, len, msg = res.unpack('CNa*')
    p Helloworld::HelloReply.decode(msg)
  end
end

session.run