Class: HTTP2::Stream

Inherits:
Object
  • Object
show all
Includes:
Emitter, Error, FlowBuffer
Defined in:
lib/http/2/stream.rb

Overview

A single HTTP 2.0 connection can multiplex multiple streams in parallel: multiple requests and responses can be in flight simultaneously and stream data can be interleaved and prioritized.

This class encapsulates all of the state, transition, flow-control, and error management as defined by the HTTP 2.0 specification. All you have to do is subscribe to appropriate events (marked with “:” prefix in diagram below) and provide your application logic to handle request and response processing.

                      +--------+
                 PP   |        |   PP
             ,--------|  idle  |--------.
            /         |        |         \
           v          +--------+          v
    +----------+          |           +----------+
    |          |          | H         |          |
,---|:reserved |          |           |:reserved |---.
|   | (local)  |          v           | (remote) |   |
|   +----------+      +--------+      +----------+   |
|      | :active      |        |      :active |      |
|      |      ,-------|:active |-------.      |      |
|      | H   /   ES   |        |   ES   \   H |      |
|      v    v         +--------+         v    v      |
|   +-----------+          |          +-----------+  |
|   |:half_close|          |          |:half_close|  |
|   |  (remote) |          |          |  (local)  |  |
|   +-----------+          |          +-----------+  |
|        |                 v                |        |
|        |    ES/R    +--------+    ES/R    |        |
|        `----------->|        |<-----------'        |
| R                   | :close |                   R |
`-------------------->|        |<--------------------'
                      +--------+

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Emitter

#add_listener, #emit, #once

Methods included from FlowBuffer

#buffered_amount

Constructor Details

#initialize(connection: nil, id: nil, weight: 16, dependency: 0, exclusive: false, parent: nil) ⇒ Stream

Initializes new stream.

Note that you should never have to call this directly. To create a new client initiated stream, use Connection#new_stream. Similarly, Connection will emit new stream objects, when new stream frames are received.

Parameters:

  • id (Integer) (defaults to: nil)
  • weight (Integer) (defaults to: 16)
  • dependency (Integer) (defaults to: 0)
  • exclusive (Boolean) (defaults to: false)
  • window (Integer)
  • parent (Stream) (defaults to: nil) 


75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/http/2/stream.rb', line 75

def initialize(connection: nil, id: nil, weight: 16, dependency: 0, exclusive: false, parent: nil)
  @connection = connection or raise ArgumentError.new("missing mandatory argument connection")
  @id = id                 or raise ArgumentError.new("missing mandatory argument id")
  @weight = weight
  @dependency = dependency
  process_priority({weight: weight, stream_dependency: dependency, exclusive: exclusive})
  @remote_window = connection.remote_settings[:settings_initial_window_size]
  @parent = parent
  @state  = :idle
  @error  = false
  @closed = false
  @send_buffer = []

  on(:window) { |v| @remote_window = v }
  on(:local_window) { |v| @local_window = v }
end

Instance Attribute Details

#closed ⇒ Object (readonly)

Reason why connection was closed.



61
62
63
 
# File 'lib/http/2/stream.rb', line 61

def closed
  @closed
end

#dependency ⇒ Object (readonly)

Returns the value of attribute dependency.



53
54
55
 
# File 'lib/http/2/stream.rb', line 53

def dependency
  @dependency
end

#id ⇒ Object (readonly)

Stream ID (odd for client initiated streams, even otherwise).



43
44
45
 
# File 'lib/http/2/stream.rb', line 43

def id
  @id
end

#local_window ⇒ Object (readonly) Also known as: window

Size of current stream flow control window.



56
57
58
 
# File 'lib/http/2/stream.rb', line 56

def local_window
  @local_window
end

#parent ⇒ Object (readonly)

Request parent stream of push stream.



49
50
51
 
# File 'lib/http/2/stream.rb', line 49

def parent
  @parent
end

#remote_window ⇒ Object (readonly)

Returns the value of attribute remote_window.



57
58
59
 
# File 'lib/http/2/stream.rb', line 57

def remote_window
  @remote_window
end

#state ⇒ Object (readonly)

Stream state as defined by HTTP 2.0.



46
47
48
 
# File 'lib/http/2/stream.rb', line 46

def state
  @state
end

#weight ⇒ Object (readonly)

Stream priority as set by initiator.



52
53
54
 
# File 'lib/http/2/stream.rb', line 52

def weight
  @weight
end

Instance Method Details

#cancel ⇒ Object

Sends a RST_STREAM indicating that the stream is no longer needed.



195
196
197
 
# File 'lib/http/2/stream.rb', line 195

def cancel
  send({type: :rst_stream, error: :cancel})
end

#close(error = :stream_closed) ⇒ Object

Sends a RST_STREAM frame which closes current stream - this does not close the underlying connection.

Parameters:

  • error (:Symbol) (defaults to: :stream_closed) —

    optional reason why stream was closed



190
191
192
 
# File 'lib/http/2/stream.rb', line 190

def close(error = :stream_closed)
  send({type: :rst_stream, error: error})
end

#data(payload, end_stream: true) ⇒ Object

Sends DATA frame containing response payload.

Parameters:

  • payload (String)
  • end_stream (Boolean) (defaults to: true) —

    indicates last response DATA frame



171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/http/2/stream.rb', line 171

def data(payload, end_stream: true)
  flags = []
  flags << :end_stream if end_stream

  # Split data according to each frame is smaller enough
  # TODO: consider padding?
  max_size = @connection.remote_settings[:settings_max_frame_size]
  while payload.bytesize > max_size do
    chunk = payload.slice!(0, max_size)
    send({type: :data, payload: chunk})
  end

  send({type: :data, flags: flags, payload: payload})
end

#headers(headers, end_headers: true, end_stream: false) ⇒ Object

Sends a HEADERS frame containing HTTP response headers.

Parameters:

  • headers (Array or Hash) —

    Array of key-value pairs or Hash

  • end_headers (Boolean) (defaults to: true) —

    indicates that no more headers will be sent

  • end_stream (Boolean) (defaults to: false) —

    indicates that no payload will be sent



142
143
144
145
146
147
148
 
# File 'lib/http/2/stream.rb', line 142

def headers(headers, end_headers: true, end_stream: false)
  flags = []
  flags << :end_headers if end_headers
  flags << :end_stream  if end_stream

  send({type: :headers, flags: flags, payload: headers.to_a})
end

#promise(headers, end_headers: true, &block) ⇒ Object

Raises:

  • (Exception) 


150
151
152
153
154
155
 
# File 'lib/http/2/stream.rb', line 150

def promise(headers, end_headers: true, &block)
  raise Exception.new("must provide callback") if !block_given?

  flags = end_headers ? [:end_headers] : []
  emit(:promise, self, headers, flags, &block)
end

#receive(frame) ⇒ Object Also known as: <<

Processes incoming HTTP 2.0 frames. The frames must be decoded upstream.

Parameters:

  • frame (Hash) 


95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/http/2/stream.rb', line 95

def receive(frame)
  transition(frame, false)

  case frame[:type]
  when :data
    # TODO: when receiving DATA, keep track of local_window.
    emit(:data, frame[:payload]) if !frame[:ignore]
  when :headers, :push_promise
    emit(:headers, frame[:payload]) if !frame[:ignore]
  when :priority
    process_priority(frame)
  when :window_update
    @remote_window += frame[:increment]
    send_data
  when :altsvc, :blocked
    emit(frame[:type], frame)
  end

  complete_transition(frame)
end

#refuse ⇒ Object

Sends a RST_STREAM indicating that the stream has been refused prior to performing any application processing.



201
202
203
 
# File 'lib/http/2/stream.rb', line 201

def refuse
  send({type: :rst_stream, error: :refused_stream})
end

#reprioritize(weight: 16, dependency: 0, exclusive: false) ⇒ Object

Sends a PRIORITY frame with new stream priority value (can only be performed by the client).

Parameters:

  • weight (Integer) (defaults to: 16) —

    new stream weight value

  • dependency (Integer) (defaults to: 0) —

    new stream dependency stream



162
163
164
165
 
# File 'lib/http/2/stream.rb', line 162

def reprioritize(weight: 16, dependency: 0, exclusive: false)
  stream_error if @id.even?
  send({type: :priority, weight: weight, stream_dependency: dependency, exclusive: exclusive})
end

#send(frame) ⇒ Object

Processes outgoing HTTP 2.0 frames. Data frames may be automatically split and buffered based on maximum frame size and current stream flow control window size.

Parameters:

  • frame (Hash) 


122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/http/2/stream.rb', line 122

def send(frame)
  transition(frame, true)
  frame[:stream] ||= @id

  process_priority(frame) if frame[:type] == :priority

  if frame[:type] == :data
    send_data(frame)
  else
    emit(:frame, frame)
  end

  complete_transition(frame)
end