Class: Raptor::DetachedBody

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/detached_body.rb,
sig/generated/raptor/detached_body.rbs

Overview

A bounded response body that may be written after its application thread has returned.

Defined Under Namespace

Classes: Budget

Constant Summary collapse

DEFAULT_MAX_BUFFER_SIZE =

Returns:

  • (Object) 
64 * 1024

Instance Method Summary collapse

Constructor Details

#initialize(max_buffer_size: DEFAULT_MAX_BUFFER_SIZE) ⇒ DetachedBody

Creates a detached body with a bounded byte buffer.

Parameters:

  • max_buffer_size (Integer) (defaults to: DEFAULT_MAX_BUFFER_SIZE) —

    maximum body bytes waiting to be written

  • max_buffer_size: (Integer) (defaults to: DEFAULT_MAX_BUFFER_SIZE) 


71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/raptor/detached_body.rb', line 71

def initialize(max_buffer_size: DEFAULT_MAX_BUFFER_SIZE)
  raise ArgumentError, "max_buffer_size must be positive" unless max_buffer_size.positive?

  @max_buffer_size = max_buffer_size
  @state = Atom.new({ chunks: [], size: 0, closing: false, closed: false, notified: false, trailers: {} })
  @budgets = []
  @wake = proc {}
  @dispatch = proc { |callback| callback.call }
  @finished = proc {}
  @on_open = nil
  @on_close = nil
end

Instance Method Details

#attach(budgets, wake, dispatch, finished) ⇒ Boolean

Connects the body to its reactor-owned response stream.

Parameters:

  • budgets (Array<Budget>) —

    buffer budgets shared with other bodies

  • wake (Proc) —

    called when buffered data or a close is ready to write

  • dispatch (Proc) —

    schedules application callbacks off the reactor thread

  • finished (Proc) —

    called with the close reason after on_close

Returns:

  • (Boolean) —

    false when the bytes already buffered exceed a budget



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/raptor/detached_body.rb', line 195

def attach(budgets, wake, dispatch, finished)
  @wake = wake
  @dispatch = dispatch
  @finished = finished
  size = @state.value[:size]
  reserved = []
  budgets.each do |budget|
    unless budget.reserve(size)
      reserved.each { |held| held.release(size) }
      finish(:full)
      return false
    end
    reserved << budget
  end

  @budgets = budgets
  true
end

#close(trailers: {}) ⇒ void

This method returns an undefined value.

Finishes the body after all accepted bytes have been written.

Parameters:

  • trailers (Hash) (defaults to: {}) —

    trailing response headers

  • trailers: (Hash[String, String | Array[String]]) (defaults to: {}) 


159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/raptor/detached_body.rb', line 159

def close(trailers: {})
  trailers = trailers.to_h do |name, value|
    value = value.is_a?(Array) ? value.map { _1.dup.freeze }.freeze : value.dup.freeze
    [name.dup.freeze, value]
  end.freeze
  wake = false
  @state.swap do |state|
    if state[:closed] || state[:closing]
      wake = false
      next state
    end

    wake = !state[:notified]
    state.merge(closing: true, notified: true, trailers: trailers)
  end
  @wake.call if wake
end

#closed? ⇒ Boolean

Returns whether the stream has closed.

Returns:

  • (Boolean) 


182
183
184
 
# File 'lib/raptor/detached_body.rb', line 182

def closed?
  @state.value[:closed]
end

#finish(reason) ⇒ void

This method returns an undefined value.

Closes the stream and invokes its callback exactly once.

Parameters:

  • reason (Symbol) —

    why the stream closed



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
 
# File 'lib/raptor/detached_body.rb', line 280

def finish(reason)
  callback = false
  remaining = 0
  @state.swap do |state|
    if state[:closed]
      callback = false
      remaining = 0
      state
    else
      callback = true
      remaining = state[:size]
      state.merge(chunks: [], size: 0, closed: true, notified: false)
    end
  end
  return unless callback

  release(remaining)
  @dispatch.call(proc do
    begin
      @on_close&.call(reason)
    ensure
      @finished.call(reason)
    end
  end)
end

#next_size ⇒ Integer?

Returns the size of the next buffered chunk, 0 when closing, or nil while waiting for more data.

Returns:

  • (Integer, nil) 


230
231
232
233
 
# File 'lib/raptor/detached_body.rb', line 230

def next_size
  state = @state.value
  state[:chunks].first&.bytesize || (0 if state[:closing])
end

#on_close {|reason| ... } ⇒ DetachedBody

Registers a callback for when the response stream closes.

Yields:

Yield Parameters:

  • reason (Symbol) —

    why the stream closed

Yield Returns:

  • (void)

Returns:



102
103
104
105
 
# File 'lib/raptor/detached_body.rb', line 102

def on_close(&block)
  @on_close = block
  self
end

#on_open {|stream| ... } ⇒ DetachedBody

Registers a callback that runs on the request thread once the server accepts the response stream.

Yields:

Yield Parameters:

Yield Returns:

  • (void)

Returns:



91
92
93
94
 
# File 'lib/raptor/detached_body.rb', line 91

def on_open(&block)
  @on_open = block
  self
end

#open ⇒ void

This method returns an undefined value.

Notifies the application that its response stream is ready.



219
220
221
222
 
# File 'lib/raptor/detached_body.rb', line 219

def open
  @on_open&.call(self)
  @wake.call if @state.value[:notified]
end

#release(bytes) ⇒ void

This method returns an undefined value.

Releases bytes from every budget attached to this body.

Parameters:

  • bytes (Integer) —

    number of bytes to release



332
333
334
 
# File 'lib/raptor/detached_body.rb', line 332

def release(bytes)
  @budgets.each { |budget| budget.release(bytes) }
end

#reserve(bytes) ⇒ Boolean

Reserves bytes from every budget attached to this body.

Parameters:

  • bytes (Integer) —

    number of bytes to reserve

Returns:

  • (Boolean) —

    whether every budget accepted the reservation



314
315
316
317
318
319
320
321
322
323
324
 
# File 'lib/raptor/detached_body.rb', line 314

def reserve(bytes)
  reserved = []
  @budgets.each do |budget|
    unless budget.reserve(bytes)
      reserved.each { |held| held.release(bytes) }
      return false
    end
    reserved << budget
  end
  true
end

#shift(max_bytes) ⇒ String?

Removes up to max_bytes from the next buffered chunk.

Parameters:

  • max_bytes (Integer) —

    the largest chunk to return

Returns:

  • (String, nil) —

    the removed bytes, or nil when nothing is buffered



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/raptor/detached_body.rb', line 241

def shift(max_bytes)
  chunk = nil
  @state.swap do |state|
    chunk = nil
    first = state[:chunks].first
    next state unless first

    size = max_bytes < first.bytesize ? max_bytes : first.bytesize
    chunk = first.byteslice(0, size)
    chunks = if size == first.bytesize
      state[:chunks].drop(1)
    else
      [first.byteslice(size..-1).freeze] + state[:chunks].drop(1)
    end
    state.merge(
      chunks: chunks,
      size: state[:size] - size,
      notified: !chunks.empty? || state[:closing]
    )
  end
  release(chunk.bytesize) if chunk
  chunk
end

#trailers ⇒ Hash

Returns the response trailers supplied when the body closed.

Returns:

  • (Hash) —

    trailing response headers



270
271
272
 
# File 'lib/raptor/detached_body.rb', line 270

def trailers
  @state.value[:trailers]
end

#try_write(chunk) ⇒ Symbol

Adds body bytes without waiting for socket or flow-control capacity.

Parameters:

  • chunk (String) —

    response body bytes

Returns:

  • (Symbol) —

    :accepted, :full, or :closed



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
 
# File 'lib/raptor/detached_body.rb', line 113

def try_write(chunk)
  raise TypeError, "body must yield String values" unless chunk.is_a?(String)

  chunk = chunk.dup.freeze
  if chunk.empty?
    state = @state.value
    return state[:closed] || state[:closing] ? :closed : :accepted
  end
  return :full unless reserve(chunk.bytesize)

  result = :closed
  wake = false
  @state.swap do |state|
    if state[:closed] || state[:closing]
      result = :closed
      wake = false
      state
    elsif state[:size] + chunk.bytesize > @max_buffer_size
      result = :full
      wake = false
      state
    else
      result = :accepted
      wake = !state[:notified]
      state.merge(
        chunks: state[:chunks] + [chunk],
        size: state[:size] + chunk.bytesize,
        notified: true
      )
    end
  end

  if result == :accepted
    @wake.call if wake
  else
    release(chunk.bytesize)
  end
  result
end