Class: Biryani::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/biryani/connection.rb

Overview

rubocop: disable Metrics/ClassLength

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(proc) ⇒ Connection

proc [Proc]



23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/biryani/connection.rb', line 23

def initialize(proc)
  @port = Ractor::Port.new
  @recv_done = false
  @proc = proc
  @streams_ctx = StreamsContext.new(proc, @port)
  @encoder = HPACK::Encoder.new(DEFAULT_HEADER_TABLE_SIZE)
  @decoder = HPACK::Decoder.new(DEFAULT_HEADER_TABLE_SIZE)
  @send_window = Window.new(INITIAL_CONNECTION_WINDOW_SIZE)
  @recv_window = Window.new(INITIAL_CONNECTION_WINDOW_SIZE)
  @data_buffer = DataBuffer.new
  @settings = self.class.default_settings # Hash<Integer, Integer>
  @peer_settings = self.class.default_settings # Hash<Integer, Integer>
  @closed = false
end

Class Method Details

.default_settings ⇒ Hash<Integer, Integer>

Returns:

  • (Hash<Integer, Integer>) 


493
494
495
496
497
498
499
500
501
502
503
 
# File 'lib/biryani/connection.rb', line 493

def self.default_settings
  # https://datatracker.ietf.org/doc/html/rfc9113#section-6.5.2
  {
    SettingsID::SETTINGS_HEADER_TABLE_SIZE => DEFAULT_HEADER_TABLE_SIZE,
    SettingsID::SETTINGS_ENABLE_PUSH => 1,
    SettingsID::SETTINGS_MAX_CONCURRENT_STREAMS => 0xffffffff,
    SettingsID::SETTINGS_INITIAL_WINDOW_SIZE => 65_535,
    SettingsID::SETTINGS_MAX_FRAME_SIZE => 16_384,
    SettingsID::SETTINGS_MAX_HEADER_LIST_SIZE => 0xffffffff
  }
end

.do_send(io, frame, flush) ⇒ Object

Parameters:

  • io (IO)
  • frame (Object)
  • flush (Boolean) 


306
307
308
309
 
# File 'lib/biryani/connection.rb', line 306

def self.do_send(io, frame, flush)
  io.write(frame.to_binary_s)
  io.flush if flush
end

.handle_connection_window_update(window_update, send_window) ⇒ nil, ConnectionError

Parameters:

  • window_update (WindowUpdate)
  • send_window (Window)

Returns:



447
448
449
450
451
452
 
# File 'lib/biryani/connection.rb', line 447

def self.handle_connection_window_update(window_update, send_window)
  send_window.increase!(window_update.window_size_increment)
  return ConnectionError.new(ErrorCode::FLOW_CONTROL_ERROR, 'flow-control window exceeds 2^31-1') if send_window.length > 2**31 - 1

  nil
end

.handle_data(stream_id, data, recv_window, streams_ctx, decoder) ⇒ Array<WindowUpdate>, ConnectionError

rubocop: disable Metrics/AbcSize

Parameters:

  • stream_id (Integer)
  • data (String)
  • recv_window (Window)
  • streams_ctx (StreamsContext)
  • decoder (Decoder)

Returns:



370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
 
# File 'lib/biryani/connection.rb', line 370

def self.handle_data(stream_id, data, recv_window, streams_ctx, decoder)
  ctx = streams_ctx[stream_id]
  return ConnectionError.new(ErrorCode::FLOW_CONTROL_ERROR, 'DATA Frame length exceeds flow-control window size') \
    if recv_window.consume!(data.bytesize).negative? || ctx.recv_window.consume!(data.bytesize).negative?

  ctx.content << data
  if ctx.half_closed_remote?
    obj = http_request(ctx.fragment, ctx.content, decoder)
    return obj if Biryani.err?(obj)

    ctx << obj
  end

  window_updates = []
  window_updates << Frame::WindowUpdate.new(0, recv_window.capacity - recv_window.length) if recv_window.length < recv_window.capacity / 2
  window_updates << Frame::WindowUpdate.new(stream_id, ctx.recv_window.capacity - ctx.recv_window.length) if ctx.recv_window.length < ctx.recv_window.capacity / 2
  window_updates
end

.handle_goaway(_goaway) ⇒ Object

Parameters:

  • _goaway (Goaway) 


441
 
# File 'lib/biryani/connection.rb', line 441

def self.handle_goaway(_goaway); end

.handle_headers(headers, ctx, decoder) ⇒ nil, ConnectionError

Parameters:

Returns:



395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/biryani/connection.rb', line 395

def self.handle_headers(headers, ctx, decoder)
  ctx.fragment << headers.fragment
  if ctx.half_closed_remote?
    obj = http_request(ctx.fragment, ctx.content, decoder)
    return obj if Biryani.err?(obj)

    ctx << obj
  end

  nil
end

.handle_ping(ping) ⇒ Ping?

Parameters:

  • ping (Ping)

Returns:

  • (Ping, nil) 


436
437
438
 
# File 'lib/biryani/connection.rb', line 436

def self.handle_ping(ping)
  Frame::Ping.new(true, 0, ping.opaque) unless ping.ack?
end

.handle_rst_stream(_rst_stream, ctx) ⇒ Object

Parameters:



409
410
411
 
# File 'lib/biryani/connection.rb', line 409

def self.handle_rst_stream(_rst_stream, ctx)
  ctx.close
end

.handle_settings(settings, peer_settings, decoder, streams_ctx) ⇒ Settings

Parameters:

  • settings (Settings)
  • peer_settings (Hash<Integer, Integer>)
  • decoder (Decoder)
  • streams_ctx (StreamsContext)

Returns:

  • (Settings) 


419
420
421
422
423
424
425
426
427
428
429
430
431
 
# File 'lib/biryani/connection.rb', line 419

def self.handle_settings(settings, peer_settings, decoder, streams_ctx)
  return nil if settings.ack?

  peer_settings.merge!(settings.setting)
  new_limit = peer_settings[SettingsID::SETTINGS_HEADER_TABLE_SIZE]
  decoder.limit!(new_limit)
  send_initial_window_size = peer_settings[SettingsID::SETTINGS_INITIAL_WINDOW_SIZE]
  streams_ctx.each do |ctx|
    ctx.send_window.update!(send_initial_window_size)
  end

  Frame::Settings.new(true, 0, {})
end

.handle_stream_window_update(window_update, streams_ctx) ⇒ nil, StreamError

Parameters:

Returns:



458
459
460
461
462
463
464
 
# File 'lib/biryani/connection.rb', line 458

def self.handle_stream_window_update(window_update, streams_ctx)
  stream_id = window_update.stream_id
  streams_ctx[stream_id].send_window.increase!(window_update.window_size_increment)
  return StreamError.new(ErrorCode::FLOW_CONTROL_ERROR, stream_id, 'flow-control window exceeds 2^31-1') if streams_ctx[stream_id].send_window.length > 2**31 - 1

  nil
end

.http_request(fragment, content, decoder) ⇒ HTTP::Request, ConnectionError

Parameters:

  • fragment (String)
  • content (String)
  • decoder (Decoder)

Returns:



471
472
473
474
475
476
477
478
479
480
481
 
# File 'lib/biryani/connection.rb', line 471

def self.http_request(fragment, content, decoder)
  obj = decoder.decode(fragment)
  return obj if Biryani.err?(obj)

  fields = obj
  builder = HTTP::RequestBuilder.new
  err = builder.fields(fields)
  return err unless err.nil?

  builder.build(content)
end

.http_response(res, encoder) ⇒ String

Parameters:

Returns:

  • (String) —

    fragment

  • (String) —

    data



488
489
490
 
# File 'lib/biryani/connection.rb', line 488

def self.http_response(res, encoder)
  HTTP::ResponseParser.new(res).parse(encoder)
end

.read_http2_magic(io) ⇒ nil, Error

Parameters:

  • io (IO)

Returns:

  • (nil, Error) 


357
358
359
360
 
# File 'lib/biryani/connection.rb', line 357

def self.read_http2_magic(io)
  s = io.read(CONNECTION_PREFACE_LENGTH)
  ConnectionError.new(ErrorCode::PROTOCOL_ERROR, 'invalid connection preface') if s != CONNECTION_PREFACE
end

.send_data(io, stream_id, data, send_window, max_frame_size, streams_ctx, data_buffer) ⇒ Object

Parameters:



318
319
320
321
322
323
324
325
326
327
328
329
 
# File 'lib/biryani/connection.rb', line 318

def self.send_data(io, stream_id, data, send_window, max_frame_size, streams_ctx, data_buffer)
  frames, remains = streams_ctx.sendable_datas(stream_id, data, send_window, max_frame_size)

  frames.each do |frame|
    do_send(io, frame, false)
    send_window.consume!(frame.length)
    streams_ctx[stream_id].send_window.consume!(frame.length)
    transition_stream_state_send(frame, streams_ctx)
  end

  data_buffer.store(stream_id, remains) unless remains.empty?
end

.send_headers(io, stream_id, fragment, only_headers, max_frame_size, streams_ctx) ⇒ Object

Parameters:

  • io (IO)
  • stream_id (Integer)
  • fragment (String)
  • only_headers (Boolean)
  • max_frame_size (Integer)
  • streams_ctx (StreamsContext) 


337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
 
# File 'lib/biryani/connection.rb', line 337

def self.send_headers(io, stream_id, fragment, only_headers, max_frame_size, streams_ctx)
  len = (fragment.bytesize + max_frame_size - 1) / max_frame_size
  frames = fragment.gsub(/.{1,#{max_frame_size}}/m).with_index.map do |s, index|
    end_headers = index == len - 1
    if index.zero?
      Frame::Headers.new(end_headers, only_headers, stream_id, nil, nil, s, nil)
    else
      Frame::Continuation.new(end_headers, stream_id, s)
    end
  end

  frames.each do |frame|
    do_send(io, frame, false)
    transition_stream_state_send(frame, streams_ctx)
  end
end

.transition_stream_state_recv(recv_frame, streams_ctx, data_buffer, stream_id, max_streams, send_initial_window_size, recv_initial_window_size) ⇒ StreamContext, ...

rubocop: disable Metrics/CyclomaticComplexity rubocop: disable Metrics/PerceivedComplexity

Parameters:

  • recv_frame (Object)
  • streams_ctx (StreamsContext)
  • stream_id (Integer)
  • data_buffer (DataBuffer)
  • max_streams (Integer)
  • send_initial_window_size (Integer)
  • recv_initial_window_size (Integer)

Returns:



264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'lib/biryani/connection.rb', line 264

def self.transition_stream_state_recv(recv_frame, streams_ctx, data_buffer, stream_id, max_streams, send_initial_window_size, recv_initial_window_size)
  ctx = streams_ctx[stream_id]
  return StreamError.new(ErrorCode::PROTOCOL_ERROR, stream_id, 'exceed max concurrent streams') if ctx.nil? && streams_ctx.count_active + 1 > max_streams
  return ConnectionError.new(ErrorCode::PROTOCOL_ERROR, 'even-numbered stream identifier') if ctx.nil? && stream_id.even?
  return ConnectionError.new(ErrorCode::PROTOCOL_ERROR, 'new stream identifier is less than the existing stream identifiers') if ctx.nil? && streams_ctx.last_stream_id > stream_id

  if ctx.nil?
    ctx = streams_ctx.new_context(stream_id, send_initial_window_size, recv_initial_window_size)
    # An ideal implementation would wait the RTT before removing the stream.
    streams_ctx.remove_closed(data_buffer)
  end

  obj = ctx.state_transition!(recv_frame, :recv)
  return obj if Biryani.err?(obj)

  ctx
end

.transition_stream_state_send(send_frame, streams_ctx) ⇒ Boolean

Returns should close connection?.

Parameters:

Returns:

  • (Boolean) —

    should close connection?



288
289
290
291
292
293
294
295
296
297
298
299
300
301
 
# File 'lib/biryani/connection.rb', line 288

def self.transition_stream_state_send(send_frame, streams_ctx)
  stream_id = send_frame.stream_id
  typ = send_frame.f_type
  case typ
  when FrameType::SETTINGS, FrameType::PING
    false
  when FrameType::GOAWAY
    streams_ctx.close_all
    true
  else
    streams_ctx[stream_id].state_transition!(send_frame, :send) unless stream_id.zero?
    false
  end
end

Instance Method Details

#close ⇒ Object 



244
245
246
 
# File 'lib/biryani/connection.rb', line 244

def close
  @closed = true
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


249
250
251
 
# File 'lib/biryani/connection.rb', line 249

def closed?
  @closed
end

#do_recv_dispatch(frame) ⇒ Array<Object>, ...

Returns frames or errors.

Parameters:

  • frame (Object)

Returns:



126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/biryani/connection.rb', line 126

def do_recv_dispatch(frame)
  receiving_continuation_stream_id = @streams_ctx.receiving_continuation_stream_id
  return [ConnectionError.new(ErrorCode::PROTOCOL_ERROR, "invalid frame type #{format('0x%02x', frame.f_type)} for stream identifier #{format('0x%02x', frame.stream_id)}")] \
    if !receiving_continuation_stream_id.nil? && frame.stream_id != receiving_continuation_stream_id

  if frame.stream_id.zero?
    handle_connection_frame(frame)
  else
    handle_stream_frame(frame)
  end
end

#handle_connection_frame(frame) ⇒ Array<Object>, ...

rubocop: disable Metrics/CyclomaticComplexity

Parameters:

  • frame (Object)

Returns:



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/biryani/connection.rb', line 142

def handle_connection_frame(frame)
  case frame.f_type
  when FrameType::DATA, FrameType::HEADERS, FrameType::PRIORITY, FrameType::RST_STREAM, FrameType::PUSH_PROMISE, FrameType::CONTINUATION
    [ConnectionError.new(ErrorCode::PROTOCOL_ERROR, "invalid frame type #{format('0x%02x', frame.f_type)} for stream identifier 0x00")]
  when FrameType::SETTINGS
    obj = self.class.handle_settings(frame, @peer_settings, @decoder, @streams_ctx)
    return [] if obj.nil?

    settings_ack = obj
    [settings_ack]
  when FrameType::PING
    obj = self.class.handle_ping(frame)
    return [] if obj.nil?

    ping_ack = obj
    [ping_ack]
  when FrameType::GOAWAY
    self.class.handle_goaway(frame)

    []
  when FrameType::WINDOW_UPDATE
    err = self.class.handle_connection_window_update(frame, @send_window)
    return [err] unless err.nil?

    max_frame_size = @peer_settings[SettingsID::SETTINGS_MAX_FRAME_SIZE]
    @data_buffer.take!(@send_window, @streams_ctx, max_frame_size) # return DATA Frames
  else
    # ignore UNKNOWN Frame

    []
  end
end

#handle_response(io, res, stream_id) ⇒ Object

Parameters:



237
238
239
240
241
242
 
# File 'lib/biryani/connection.rb', line 237

def handle_response(io, res, stream_id)
  fragment, data = self.class.http_response(res, @encoder)
  max_frame_size = @peer_settings[SettingsID::SETTINGS_MAX_FRAME_SIZE]
  self.class.send_headers(io, stream_id, fragment, data.empty?, max_frame_size, @streams_ctx)
  self.class.send_data(io, stream_id, data, @send_window, max_frame_size, @streams_ctx, @data_buffer) unless data.empty?
end

#handle_stream_frame(frame) ⇒ Array<Object>, ...

rubocop: disable Metrics/AbcSize rubocop: disable Metrics/CyclomaticComplexity rubocop: disable Metrics/MethodLength

Parameters:

  • frame (Object)

Returns:



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/biryani/connection.rb', line 182

def handle_stream_frame(frame)
  stream_id = frame.stream_id
  typ = frame.f_type
  return [ConnectionError.new(ErrorCode::PROTOCOL_ERROR, "invalid frame type #{format('0x%02x', typ)} for stream identifier #{format('0x%02x', stream_id)}")] \
    if [FrameType::SETTINGS, FrameType::PING, FrameType::GOAWAY].include?(typ)

  max_streams = @peer_settings[SettingsID::SETTINGS_MAX_CONCURRENT_STREAMS]
  send_initial_window_size = @peer_settings[SettingsID::SETTINGS_INITIAL_WINDOW_SIZE]
  recv_initial_window_size = @settings[SettingsID::SETTINGS_INITIAL_WINDOW_SIZE]
  obj = self.class.transition_stream_state_recv(frame, @streams_ctx, @data_buffer, stream_id, max_streams, send_initial_window_size, recv_initial_window_size)
  return [obj] if Biryani.err?(obj)

  ctx = obj
  case typ
  when FrameType::DATA
    obj = self.class.handle_data(stream_id, frame.data, @recv_window, @streams_ctx, @decoder)
    return [obj] if Biryani.err?(obj)

    obj # return WINDOW_UPDATE Frames
  when FrameType::HEADERS, FrameType::CONTINUATION
    err = self.class.handle_headers(frame, ctx, @decoder)
    return [err] unless err.nil?

    []
  when FrameType::PRIORITY
    # ignore PRIORITY Frame

    []
  when FrameType::PUSH_PROMISE
    # TODO

    []
  when FrameType::RST_STREAM
    self.class.handle_rst_stream(frame, ctx)

    []
  when FrameType::WINDOW_UPDATE
    err = self.class.handle_stream_window_update(frame, @streams_ctx)
    return [err] unless err.nil?

    max_frame_size = @peer_settings[SettingsID::SETTINGS_MAX_FRAME_SIZE]
    @data_buffer.take!(@send_window, @streams_ctx, max_frame_size)
  else
    # ignore UNKNOWN Frame

    []
  end
end

#recv_dispatch(io, obj) ⇒ Array<Object>, ...

rubocop: disable Metrics/AbcSize rubocop: disable Metrics/CyclomaticComplexity rubocop: disable Metrics/PerceivedComplexity

Parameters:

  • io (IO)
  • obj (Object)

Returns:



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/biryani/connection.rb', line 97

def recv_dispatch(io, obj)
  if Biryani.err?(obj)
    reply_frame = Biryani.unwrap(obj, @streams_ctx.last_stream_id)
    self.class.do_send(io, reply_frame, true)
    close if self.class.transition_stream_state_send(reply_frame, @streams_ctx)
  elsif obj.length > @settings[SettingsID::SETTINGS_MAX_FRAME_SIZE]
    self.class.do_send(io, Frame::Goaway.new(@streams_ctx.last_stream_id, ErrorCode::FRAME_SIZE_ERROR, 'payload length greater than SETTINGS_MAX_FRAME_SIZE'), true)
    close
  else
    do_recv_dispatch(obj).each do |frame|
      reply_frame = Biryani.unwrap(frame, @streams_ctx.last_stream_id)
      self.class.do_send(io, reply_frame, true)
      if reply_frame.f_type == FrameType::WINDOW_UPDATE && reply_frame.stream_id.zero?
        @recv_window.increase!(reply_frame.window_size_increment)
      elsif reply_frame.f_type == FrameType::WINDOW_UPDATE
        @streams_ctx[reply_frame.stream_id].recv_window.increase!(reply_frame.window_size_increment)
      end

      close if self.class.transition_stream_state_send(reply_frame, @streams_ctx)
    end
  end
end

#recv_loop(io) ⇒ Object

Parameters:

  • io (IO) 


58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/biryani/connection.rb', line 58

def recv_loop(io)
  Ractor.new(io, @port) do |io_, port_|
    loop do
      obj = Frame.read(io_)
      if obj.nil?
        port_.send(:eof, move: true)
        break
      end

      port_.send([:frame, obj], move: true)
    end
  end
end

#select_loop(io) ⇒ Object

Parameters:

  • io (IO) 


73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/biryani/connection.rb', line 73

def select_loop(io)
  loop do
    break if @recv_done && @streams_ctx.empty?

    case @port.receive
    in :eof
      @recv_done = true
    in [:frame, obj]
      recv_dispatch(io, obj)
    in [:response, res, stream_id]
      handle_response(io, res, stream_id)
    end

    break if closed?
  end
end

#serve(io) ⇒ Object

Parameters:

  • io (IO) 


39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/biryani/connection.rb', line 39

def serve(io)
  err = self.class.read_http2_magic(io)
  unless err.nil?
    self.class.do_send(io, err.goaway(@streams_ctx.last_stream_id), true)
    return
  end

  self.class.do_send(io, Frame::Settings.new(false, 0, {}), true)

  recv_loop(io.clone)
  select_loop(io)
rescue StandardError => e
  puts e.backtrace
  self.class.do_send(io, Frame::Goaway.new(@streams_ctx.last_stream_id, ErrorCode::INTERNAL_ERROR, 'internal error'), true)
ensure
  io.close_write
end