Class: Async::HTTP::Protocol::HTTP2::Stream::Buffer

Inherits:
Object
  • Object
show all
Defined in:
lib/async/http/protocol/http2/stream.rb

Instance Method Summary collapse

Constructor Details

#initialize(stream, body, task: Task.current) ⇒ Buffer

Returns a new instance of Buffer.



29
30
31
32
33
34
35
36
37
38
 
# File 'lib/async/http/protocol/http2/stream.rb', line 29

def initialize(stream, body, task: Task.current)
  @stream = stream
  
  @body = body
  @remainder = nil
  
  @window_updated = Async::Condition.new
  
  @task = task.async(&self.method(:passthrough))
end

Instance Method Details

#close(error) ⇒ Object 



100
101
102
103
104
105
106
107
 
# File 'lib/async/http/protocol/http2/stream.rb', line 100

def close(error)
  if @body
    @body.close(error)
    @body = nil
  end
  
  @task&.stop
end

#end_stream ⇒ Object 



92
93
94
 
# File 'lib/async/http/protocol/http2/stream.rb', line 92

def end_stream
  @stream.send_data(nil, ::Protocol::HTTP2::END_STREAM)
end

#passthrough(task) ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/async/http/protocol/http2/stream.rb', line 40

def passthrough(task)
  while chunk = self.read
    maximum_size = @stream.available_frame_size
    
    while maximum_size <= 0
      @window_updated.wait
      
      maximum_size = @stream.available_frame_size
    end
    
    self.send_data(chunk, maximum_size)
  end
  
  self.end_stream
rescue Async::Stop
  # Ignore.
ensure
  @body&.close($!)
  @body = nil
end

#push(chunk) ⇒ Object 



72
73
74
 
# File 'lib/async/http/protocol/http2/stream.rb', line 72

def push(chunk)
  @remainder = chunk
end

#read ⇒ Object 



61
62
63
64
65
66
67
68
69
70
 
# File 'lib/async/http/protocol/http2/stream.rb', line 61

def read
  if @remainder
    remainder = @remainder
    @remainder = nil
    
    return remainder
  else
    @body.read
  end
end

#send_data(chunk, maximum_size) ⇒ Object

Send ‘maximum_size` bytes of data using the specified `stream`. If the buffer has no more chunks, `END_STREAM` will be sent on the final chunk.

Parameters:

  • maximum_size (Integer) —

    send up to this many bytes of data.

  • stream (Stream) —

    the stream to use for sending data frames.



79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/async/http/protocol/http2/stream.rb', line 79

def send_data(chunk, maximum_size)
  if chunk.bytesize <= maximum_size
    @stream.send_data(chunk, maximum_size: maximum_size)
  else
    @stream.send_data(chunk.byteslice(0, maximum_size), maximum_size: maximum_size)
    
    # The window was not big enough to send all the data, so we save it for next time:
    self.push(
      chunk.byteslice(maximum_size, chunk.bytesize - maximum_size)
    )
  end
end

#window_updated(size) ⇒ Object 



96
97
98
 
# File 'lib/async/http/protocol/http2/stream.rb', line 96

def window_updated(size)
  @window_updated.signal
end