Class: LittleGhost::Support::InterruptibleStream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/little_ghost/support/interruptible_stream.rb

Overview

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) } }

Defined Under Namespace

Classes: CleanupError

Constant Summary collapse

POLL_INTERVAL =

:nodoc:

0.05
SHUTDOWN_TIMEOUT =

:nodoc:

0.1
BUFFER_SIZE =

:nodoc:

16

Instance Method Summary collapse

Constructor Details

#initialize(cancellation_token:, deadline: nil, buffer_size: BUFFER_SIZE, &producer) ⇒ InterruptibleStream

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

Raises:

  • (ArgumentError) 


28
29
30
31
32
33
34
35
36
 
# 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

Instance Method Details

#each ⇒ Object

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



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# 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