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)

Raises:

  • (ArgumentError) —

    if max_buffer_size is not positive



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

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



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/raptor/detached_body.rb', line 199

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.

Marks the body complete; the stream finishes once buffered bytes are written.

Parameters:

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

    trailing response headers

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


163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/raptor/detached_body.rb', line 163

def close(trailers: {})
  trailers = trailers.to_h do |name, value|
    value = value.is_a?(Array) ? value.map { |item| item.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) 


186
187
188
 
# File 'lib/raptor/detached_body.rb', line 186

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



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

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) 


234
235
236
237
 
# File 'lib/raptor/detached_body.rb', line 234

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:

  • once the response stream closes

Yield Parameters:

  • reason (Symbol) —

    why the stream closed

Returns:



105
106
107
108
 
# File 'lib/raptor/detached_body.rb', line 105

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:

  • once the response stream opens

Yield Parameters:

Returns:



93
94
95
96
 
# File 'lib/raptor/detached_body.rb', line 93

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.



223
224
225
226
 
# File 'lib/raptor/detached_body.rb', line 223

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



336
337
338
 
# File 'lib/raptor/detached_body.rb', line 336

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



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

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



245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/raptor/detached_body.rb', line 245

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



274
275
276
 
# File 'lib/raptor/detached_body.rb', line 274

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

Raises:

  • (TypeError) —

    if the chunk is not a String



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
152
153
154
155
 
# File 'lib/raptor/detached_body.rb', line 117

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