Class: Cadenya::Core::DefaultStreamTransport

Inherits:

Object

• Object
• Cadenya::Core::DefaultStreamTransport

Defined in:

lib/cadenya/core.rb

Overview

The default SSE transport: interruptible Net::HTTP. Implements the injectable stream-transport interface — stream yields body chunks for ANY status and returns [status, diagnostic_error_body_or_nil]; the cancel handle's teardown must unblock a parked read.

Instance Method Summary

• #stream(method:, uri:, headers:, body:, cancel: nil, open_timeout: 60, &on_chunk) ⇒ Object

Instance Method Details

#stream(method:, uri:, headers:, body:, cancel: nil, open_timeout: 60, &on_chunk) ⇒ Object

# File 'lib/cadenya/core.rb', line 293

def stream(method:, uri:, headers:, body:, cancel: nil, open_timeout: 60, &on_chunk)
  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = uri.scheme == "https"
  # A per-request timeout bounds only the OPEN phase for streams: a
  # deadline on a healthy quiet SSE body would be a lifetime limit.
  http.open_timeout = open_timeout
  # The HANDSHAKE (response headers) is bounded — a half-open server
  # must not hang the request forever. Only the BODY read is unbounded:
  # a healthy SSE stream may stay silent indefinitely.
  http.read_timeout = open_timeout

  req = Net::HTTP.const_get(method.to_s.capitalize).new(uri.request_uri)
  headers.each { |k, v| req[k] = v }
  req.body = body if body

  status = nil
  error_body = nil
  begin
    http.start
    # Registered only once the connection exists; a close that already
    # happened fires the teardown immediately inside attach.
    cancel&.attach do
      http.finish
    rescue IOError
      nil
    end
    http.request(req) do |response|
      status = response.code.to_i
      if (200...300).cover?(status)
        http.read_timeout = nil
        response.read_body(&on_chunk)
      else
        # Diagnostic body read bounded in BYTES and TIME: the stream's
        # read timeout was disabled before the status was known, so a
        # server that stalls after failure headers must not hang this.
        http.read_timeout = STREAM_ERROR_BODY_SECONDS
        error_body = +""
        begin
          response.read_body do |chunk|
            if error_body.bytesize < MAX_ERROR_BODY
              error_body << chunk.byteslice(0, MAX_ERROR_BODY - error_body.bytesize)
            end
            break if error_body.bytesize >= MAX_ERROR_BODY
          end
        rescue Net::ReadTimeout, IOError, EOFError
          # Partial diagnostics still name the failure.
        end
      end
    end
  ensure
    begin
      http.finish if http.started?
    rescue IOError
      nil
    end
  end
  [status, error_body]
end