class LittleGhost::Support::InterruptibleStream

InterruptibleStream turns a blocking producer into a lazy, cancellable Ruby stream. It is useful when an SDK owns the blocking read but the agent still needs deadlines and cooperative cancellation.

The producer receives an emitter callable. Ending enumeration early stops and joins the producer; CleanupError is raised if it cannot be stopped within the fixed shutdown bound.

stream = LittleGhost::Support::InterruptibleStream.new(
  cancellation_token: token
) { |emit| source.each { |value| emit.call(value) } }

Public Class Methods

new (cancellation_token:, deadline: nil, buffer_size: BUFFER_SIZE, &producer)

Source

# File lib/little_ghost/support/interruptible_stream.rb, line 28
def initialize(cancellation_token:, deadline: nil, buffer_size: BUFFER_SIZE, &producer)
  raise ArgumentError, "producer is required" unless producer

  @cancellation_token = cancellation_token
  @deadline = deadline
  @buffer_size = Integer(buffer_size)
  @producer = producer
  raise ArgumentError, "buffer_size must be positive" unless @buffer_size.positive?
end

Configures a lazy stream. The producer starts when each is consumed.

Public Instance Methods

each () { |value| ... }

Source

# File lib/little_ghost/support/interruptible_stream.rb, line 40
def each
  return enum_for(__method__) unless block_given?

  queue = SizedQueue.new(@buffer_size)
  execution_state = ExecutionState.capture
  worker = Thread.new do
    ExecutionState.with(execution_state) do
      @producer.call(->(value) { queue << [:value, value] })
    end
  rescue => error
    queue << [:error, error]
  end
  worker.report_on_exception = false

  loop do
    check!
    break if !worker.alive? && queue.empty?

    item = next_item(queue)
    unless item
      break if !worker.alive? && queue.empty?
      next
    end

    type, value = item
    check!
    case type
    when :value then yield value
    when :error then raise value
    end
  end
  self
ensure
  if worker
    worker.kill if worker.alive?
    worker.join(SHUTDOWN_TIMEOUT)
    if worker.alive?
      raise CleanupError,
        "stream producer did not stop within #{SHUTDOWN_TIMEOUT} seconds"
    end
  end
end

Yields produced values, raising producer, cancellation, deadline, or cleanup errors in the consuming thread.