Class: Taski::Execution::TaskOutputRouter

Inherits:
Object
  • Object
show all
Includes:
MonitorMixin
Defined in:
lib/taski/execution/task_output_router.rb

Overview

Central coordinator that manages all task pipes and polling. Also acts as an IO proxy for $stdout, routing writes to the appropriate pipe based on the current thread.

Architecture:

  • Each task gets a dedicated IO pipe for output capture
  • Writes are routed to the appropriate pipe based on Thread.current
  • A reader thread polls all pipes using IO.select for efficiency
  • When no pipe is registered for a thread, output goes to original stdout

Constant Summary collapse

POLL_TIMEOUT =

50ms timeout for IO.select

0.05
POLL_INTERVAL =

100ms between polls (matches TreeProgressDisplay)

0.1
READ_BUFFER_SIZE = 
4096
MAX_RECENT_LINES =

Maximum number of recent lines to keep per task

30

Instance Method Summary collapse

Constructor Details

#initialize(original_stdout, execution_facade = nil) ⇒ TaskOutputRouter

Returns a new instance of TaskOutputRouter.



25
26
27
28
29
30
31
32
33
34
 
# File 'lib/taski/execution/task_output_router.rb', line 25

def initialize(original_stdout, execution_facade = nil)
  super()
  @original = original_stdout
  @execution_facade = execution_facade
  @pipes = {}         # task_class => TaskOutputPipe
  @thread_map = {}    # Thread => task_class
  @recent_lines = {}  # task_class => Array<String>
  @poll_thread = nil
  @polling = false
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(method) ⇒ Object 



201
202
203
 
# File 'lib/taski/execution/task_output_router.rb', line 201

def method_missing(method, ...)
  @original.send(method, ...)
end

Instance Method Details

#<<(str) ⇒ Object 



169
170
171
172
 
# File 'lib/taski/execution/task_output_router.rb', line 169

def <<(str)
  write(str.to_s)
  self
end

#active? ⇒ Boolean

Returns:

  • (Boolean) 


129
130
131
132
133
 
# File 'lib/taski/execution/task_output_router.rb', line 129

def active?
  synchronize do
    @pipes.values.any? { |p| !p.read_closed? }
  end
end

#close_all ⇒ Object 



121
122
123
124
125
126
127
 
# File 'lib/taski/execution/task_output_router.rb', line 121

def close_all
  synchronize do
    @pipes.each_value(&:close)
    @pipes.clear
    @thread_map.clear
  end
end

#current_write_io ⇒ Object

Used by Task#system to redirect subprocess output to the pipe.



191
192
193
194
195
196
197
198
199
 
# File 'lib/taski/execution/task_output_router.rb', line 191

def current_write_io
  synchronize do
    task_class = @thread_map[Thread.current]
    return nil unless task_class
    pipe = @pipes[task_class]
    return nil if pipe.nil? || pipe.write_closed?
    pipe.write_io
  end
end

#flush ⇒ Object 



174
175
176
 
# File 'lib/taski/execution/task_output_router.rb', line 174

def flush
  @original.flush
end

#isatty ⇒ Object 



182
183
184
 
# File 'lib/taski/execution/task_output_router.rb', line 182

def isatty
  @original.isatty
end

#last_line_for(task_class) ⇒ Object 



110
111
112
 
# File 'lib/taski/execution/task_output_router.rb', line 110

def last_line_for(task_class)
  synchronize { @recent_lines[task_class]&.last }
end

#poll ⇒ Object

Called periodically from the display thread.



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/taski/execution/task_output_router.rb', line 90

def poll
  readable_pipes = synchronize do
    @pipes.values.reject { |p| p.read_closed? }.map(&:read_io)
  end
  return if readable_pipes.empty?

  # Handle race condition: pipe may be closed between check and select
  ready, = IO.select(readable_pipes, nil, nil, POLL_TIMEOUT)
  return unless ready

  ready.each do |read_io|
    pipe = synchronize { @pipes.values.find { |p| p.read_io == read_io } }
    next unless pipe

    read_from_pipe(pipe)
  end
rescue IOError, Errno::EBADF
  # Pipe was closed by another thread (drain_pipe), ignore
end

#print(*args) ⇒ Object 



164
165
166
167
 
# File 'lib/taski/execution/task_output_router.rb', line 164

def print(*args)
  args.each { |arg| write(arg.to_s) }
  nil
end

#puts(*args) ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/taski/execution/task_output_router.rb', line 151

def puts(*args)
  if args.empty?
    write("\n")
  else
    args.each do |arg|
      str = arg.to_s
      write(str)
      write("\n") unless str.end_with?("\n")
    end
  end
  nil
end

#read(task_class, limit: nil) ⇒ Object 



114
115
116
117
118
119
 
# File 'lib/taski/execution/task_output_router.rb', line 114

def read(task_class, limit: nil)
  synchronize do
    lines = (@recent_lines[task_class] || []).dup
    limit ? lines.last(limit) : lines
  end
end

#respond_to_missing?(method, include_private = false) ⇒ Boolean

Returns:

  • (Boolean) 


205
206
207
 
# File 'lib/taski/execution/task_output_router.rb', line 205

def respond_to_missing?(method, include_private = false)
  @original.respond_to?(method, include_private)
end

#start_capture(task_class) ⇒ Object 



59
60
61
62
63
64
65
66
 
# File 'lib/taski/execution/task_output_router.rb', line 59

def start_capture(task_class)
  synchronize do
    pipe = TaskOutputPipe.new(task_class)
    @pipes[task_class] = pipe
    @thread_map[Thread.current] = task_class
    Taski::Logging.debug(Taski::Logging::Events::OUTPUT_ROUTER_START_CAPTURE, task: task_class.name)
  end
end

#start_polling ⇒ Object

Start the background polling thread This ensures pipes are drained even when display doesn't poll



38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/taski/execution/task_output_router.rb', line 38

def start_polling
  synchronize do
    return if @polling
    @polling = true
  end

  @poll_thread = Thread.new do
    loop do
      break unless @polling
      poll
      sleep POLL_INTERVAL
    end
  end
end

#stop_capture ⇒ Object

Closes the write end and drains remaining data.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/taski/execution/task_output_router.rb', line 69

def stop_capture
  task_class = nil
  pipe = nil

  synchronize do
    task_class = @thread_map.delete(Thread.current)
    unless task_class
      Taski::Logging.debug(Taski::Logging::Events::OUTPUT_ROUTER_STOP_CAPTURE_UNREGISTERED)
      return
    end

    pipe = @pipes[task_class]
    pipe&.close_write
    Taski::Logging.debug(Taski::Logging::Events::OUTPUT_ROUTER_STOP_CAPTURE, task: task_class.name)
  end

  # Drain any remaining data from the pipe after closing write end
  drain_pipe(pipe) if pipe
end

#stop_polling ⇒ Object 



53
54
55
56
57
 
# File 'lib/taski/execution/task_output_router.rb', line 53

def stop_polling
  synchronize { @polling = false }
  @poll_thread&.join(0.5)
  @poll_thread = nil
end

#tty? ⇒ Boolean

Returns:

  • (Boolean) 


178
179
180
 
# File 'lib/taski/execution/task_output_router.rb', line 178

def tty?
  @original.tty?
end

#winsize ⇒ Object 



186
187
188
 
# File 'lib/taski/execution/task_output_router.rb', line 186

def winsize
  @original.winsize
end

#write(str) ⇒ Object

IO interface methods - route to pipe when capturing, otherwise pass through



137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/taski/execution/task_output_router.rb', line 137

def write(str)
  pipe = current_thread_pipe
  if pipe && !pipe.write_closed?
    begin
      pipe.write_io.write(str)
    rescue IOError
      # Pipe was closed by another thread (e.g., stop_capture), fall back to original
      @original.write(str)
    end
  else
    @original.write(str)
  end
end