Class: Tina4Ultipa::GrpcClient::Channel

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4_ultipa/grpc.rb

Overview

A single HTTP/2 connection to a gqldb host. Not thread-safe; use one Channel per Client instance.

Constant Summary collapse

READ_CHUNK = 
16_384

Instance Method Summary collapse

Constructor Details

#initialize(host, port, use_tls: false, connect_timeout: 10.0) ⇒ Channel

Returns a new instance of Channel.



53
54
55
56
57
58
59
60
61
 
# File 'lib/tina4_ultipa/grpc.rb', line 53

def initialize(host, port, use_tls: false, connect_timeout: 10.0)
  @host = host
  @port = port.to_i
  @use_tls = use_tls
  @connect_timeout = connect_timeout.to_f
  @socket = nil
  @client = nil
  @closed = false
end

Instance Method Details

#close ⇒ Object 



164
165
166
167
168
169
170
171
 
# File 'lib/tina4_ultipa/grpc.rb', line 164

def close
  return if @closed

  @closed = true
  @socket&.close rescue nil
  @socket = nil
  @client = nil
end

#connect ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/tina4_ultipa/grpc.rb', line 63

def connect
  return self if @socket

  deadline = monotonic + @connect_timeout

  # Plain TCP connect with a wall-clock deadline. Socket.tcp handles
  # IPv4/IPv6 fallback; a bare TCPSocket does not.
  begin
    @socket = Socket.tcp(@host, @port, connect_timeout: @connect_timeout)
  rescue Errno::ETIMEDOUT, Errno::ECONNREFUSED, Errno::EHOSTUNREACH,
         Errno::ENETUNREACH, SocketError => e
    raise ConnectTimeout, "tcp: #{e.class}: #{e.message}"
  end

  if @use_tls
    ctx = OpenSSL::SSL::SSLContext.new
    ctx.alpn_protocols = ["h2"]
    @socket = OpenSSL::SSL::SSLSocket.new(@socket, ctx)
    @socket.sync_close = true
    @socket.hostname = @host
    begin
      @socket.connect
    rescue OpenSSL::SSL::SSLError => e
      raise ConnectTimeout, "tls: #{e.message}"
    end
    if @socket.respond_to?(:alpn_protocol) && @socket.alpn_protocol != "h2"
      raise ConnectTimeout, "alpn: server did not select h2"
    end
  end

  # http-2's Client emits :frame events with encoded HTTP/2 frames.
  # Write them straight to the socket. All inbound bytes get fed back
  # via `client << data`, which raises the state-machine events on
  # the stream we open. The client preface + initial SETTINGS fires
  # automatically the first time we emit a frame in `unary`, so there
  # is nothing to do here beyond having a live socket + a Client
  # instance ready to receive callbacks.
  @client = HTTP2::Client.new
  @client.on(:frame) { |bytes| @socket.write(bytes) }
  self
end

#unary(method, payload, metadata: {}, timeout: nil) ⇒ Object

unary(method, payload_bytes, metadata:, timeout:) -> response payload bytes

Raises RpcError on non-OK grpc-status, ConnectTimeout on connect/read deadline overrun.



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/tina4_ultipa/grpc.rb', line 109

def unary(method, payload, metadata: {}, timeout: nil)
  connect
  deadline = timeout ? monotonic + timeout.to_f : nil

  headers = {
    ":method" => "POST",
    ":scheme" => @use_tls ? "https" : "http",
    ":path" => method,
    ":authority" => "#{@host}:#{@port}",
    "content-type" => "application/grpc+proto",
    "te" => "trailers",
    "user-agent" => "tina4-ultipa-ruby/0.2 http-2",
  }
  metadata.each { |k, v| headers[k.to_s.downcase] = v.to_s }

  stream = @client.new_stream

  resp_status = nil
  resp_message = nil
  resp_body = "".b
  stream_closed = false
  rst_reason = nil

  stream.on(:headers) do |h|
    h.each do |k, v|
      case k
      when "grpc-status"  then resp_status = v.to_i
      when "grpc-message" then resp_message = v
      end
    end
  end
  stream.on(:data) { |d| resp_body << d.b }
  stream.on(:close) { stream_closed = true }
  stream.on(:reset) { |code| rst_reason = code; stream_closed = true }

  stream.headers(headers, end_stream: false)
  stream.data(frame_encode(payload), end_stream: true)

  drain_until(deadline) { stream_closed }

  if rst_reason
    raise RpcError.new(2, "stream reset (#{rst_reason})")
  end
  # No grpc-status trailer at all is a protocol violation (or the server
  # closed the connection); surface as UNKNOWN so the caller sees why.
  if resp_status.nil?
    raise RpcError.new(2, "no grpc-status trailer (server may have closed)")
  end
  unless resp_status.zero?
    raise RpcError.new(resp_status, resp_message || "")
  end

  frame_decode(resp_body)
end