module HTTPX::Plugins::StreamBidi::RequestMethods

  1. lib/httpx/plugins/stream_bidi.rb

Adds synchronization to request operations which may buffer payloads from different threads.

Attributes

headers_sent [RW]

Public Class methods

new(*)
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
261 def initialize(*)
262   super
263   @headers_sent = false
264   @closed = false
265   @flush_buffer_on_body_cb = nil
266   @mutex = Thread::Mutex.new
267 end

Public Instance methods

<<(chunk)
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
319 def <<(chunk)
320   @mutex.synchronize do
321     if @drainer
322       @body.clear if @body.respond_to?(:clear)
323       @drainer = nil
324     end
325     @body << chunk
326 
327     transition(:body)
328   end
329 end
can_buffer?()
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
279 def can_buffer?
280   return super unless @options.stream
281 
282   super && @state != :waiting_for_chunk
283 end
close()
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
331 def close
332   return super unless @options.stream
333 
334   @mutex.synchronize do
335     return if @closed
336 
337     @closed = true
338   end
339 
340   # last chunk to send which ends the stream
341   self << ""
342 end
closed?()
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
273 def closed?
274   return super unless @options.stream
275 
276   @closed
277 end
flush_buffer_on_body(&cb)
[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
269 def flush_buffer_on_body(&cb)
270   @flush_buffer_on_body_cb = on(:body, &cb)
271 end
transition(nextstate)

overrides state management transitions to introduce an intermediate :waiting_for_chunk state, which the request transitions to once payload is buffered.

[show source] 
    # File lib/httpx/plugins/stream_bidi.rb
288 def transition(nextstate)
289   return super unless @options.stream
290 
291   headers_sent = @headers_sent
292 
293   case nextstate
294   when :idle
295     headers_sent = false
296 
297     if @flush_buffer_on_body_cb
298       callbacks(:body).delete(@flush_buffer_on_body_cb)
299       @flush_buffer_on_body_cb = nil
300     end
301   when :waiting_for_chunk
302     return unless @state == :body
303   when :body
304     case @state
305     when :headers
306       headers_sent = true
307     when :waiting_for_chunk
308       # HACK: to allow super to pass through
309       @state = :headers
310     end
311   end
312 
313   super.tap do
314     # delay setting this up until after the first transition to :body
315     @headers_sent = headers_sent
316   end
317 end