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

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delegate, *args) ⇒ Stream

Returns a new instance of Stream.



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

def initialize(delegate, *args)
  super(*args)
  
  @delegate = delegate
  
  # This is the body that is being sent.
  @body = nil
  
  # The remainder of the current chunk being sent.
  @remainder = nil
  
  # The task that is handling sending the body.
  @task = nil
end

Instance Attribute Details

#body ⇒ Object (readonly)

Returns the value of attribute body.



44
45
46
 
# File 'lib/async/http/protocol/http2/stream.rb', line 44

def body
  @body
end

#delegate ⇒ Object

Returns the value of attribute delegate.



43
44
45
 
# File 'lib/async/http/protocol/http2/stream.rb', line 43

def delegate
  @delegate
end

Instance Method Details

#accept_push_promise_stream(headers, stream_id) ⇒ Object 



50
51
52
 
# File 'lib/async/http/protocol/http2/stream.rb', line 50

def accept_push_promise_stream(headers, stream_id)
  @delegate.accept_push_promise_stream(headers, stream_id)
end

#close(error = nil) ⇒ Object 



149
150
151
152
153
154
155
156
157
158
 
# File 'lib/async/http/protocol/http2/stream.rb', line 149

def close(error = nil)
  super
  
  if @body
    @body.close(error)
    @body = nil
  end
  
  @delegate.stream_closed(error)
end

#close! ⇒ Object 



143
144
145
146
147
 
# File 'lib/async/http/protocol/http2/stream.rb', line 143

def close!
  @delegate.close!
  
  super
end

#create_push_promise_stream(headers) ⇒ Object 



46
47
48
 
# File 'lib/async/http/protocol/http2/stream.rb', line 46

def create_push_promise_stream(headers)
  @delegate.create_push_promise_stream(headers)
end

#receive_data(frame) ⇒ Object 



120
121
122
123
124
125
126
127
128
 
# File 'lib/async/http/protocol/http2/stream.rb', line 120

def receive_data(frame)
  data = super
  
  if data
    @delegate.receive_data(self, data, frame.end_stream?)
  end
  
  return data
end

#receive_headers(frame) ⇒ Object 



112
113
114
115
116
117
118
 
# File 'lib/async/http/protocol/http2/stream.rb', line 112

def receive_headers(frame)
  headers = super
  
  @delegate.receive_headers(self, headers, frame.end_stream?)
  
  return headers
end

#receive_reset_stream(frame) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/async/http/protocol/http2/stream.rb', line 130

def receive_reset_stream(frame)
  error_code = super
  
  if @body
    @body.close(EOFError.new(error_code))
    @body = nil
  end
  
  @delegate.receive_reset_stream(self, error_code)
  
  return error_code
end

#send_body(body, task: Async::Task.current) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
 
# File 'lib/async/http/protocol/http2/stream.rb', line 54

def send_body(body, task: Async::Task.current)
  # TODO Might need to stop this task when body is cancelled.
  @task = task.async do |subtask|
    subtask.annotate "Sending body: #{body.class}"
    
    @body = body
    
    window_updated
  end
end

#send_chunk ⇒ Object 



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
 
# File 'lib/async/http/protocol/http2/stream.rb', line 65

def send_chunk
  maximum_size = self.available_frame_size
  
  if maximum_size == 0
    return false
  end
  
  if @remainder
    chunk = @remainder
    @remainder = nil
  elsif chunk = @body.read
    # There was a new chunk of data to send
  else
    if @body
      @body.close
      @body = nil
    end
    
    # @body.read above might take a while and a stream reset might be received in the mean time.
    unless closed? or @connection.closed?
      send_data(nil, ::Protocol::HTTP2::END_STREAM)
    end
    
    return false
  end
  
  return false if closed?
  
  if chunk.bytesize <= maximum_size
    send_data(chunk, maximum_size: maximum_size)
  else
    send_data(chunk.byteslice(0, maximum_size), maximum_size: maximum_size)
    
    @remainder = chunk.byteslice(maximum_size, chunk.bytesize - maximum_size)
  end
  
  return true
end

#window_updated ⇒ Object 



104
105
106
107
108
109
110
 
# File 'lib/async/http/protocol/http2/stream.rb', line 104

def window_updated
  return unless @body
  
  while send_chunk
    # There could be more data to send...
  end
end