Class: Fractor::ContinuousServer

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/continuous_server.rb

Overview

High-level wrapper for running Fractor in continuous mode. Handles threading, signal handling, and results processing automatically.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_pools:, work_queue: nil, log_file: nil, logger: nil) ⇒ ContinuousServer

Initialize a continuous server

Parameters:

  • worker_pools (Array<Hash>) —

    Worker pool configurations

  • work_queue (WorkQueue, nil) (defaults to: nil) —

    Optional work queue to auto-register

  • log_file (String, nil) (defaults to: nil) —

    Optional log file path

  • logger (Logger, nil) (defaults to: nil) —

    Optional logger instance for isolation (defaults to Fractor.logger)



16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/fractor/continuous_server.rb', line 16

def initialize(worker_pools:, work_queue: nil, log_file: nil, logger: nil)
  @worker_pools = worker_pools
  @work_queue = work_queue
  @log_file_path = log_file
  @log_file = nil
  @logger = logger # Store instance-specific logger for isolation
  @result_callbacks = []
  @error_callbacks = []
  @supervisor = nil
  @supervisor_thread = nil
  @running = false
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



9
10
11
 
# File 'lib/fractor/continuous_server.rb', line 9

def logger
  @logger
end

#supervisor ⇒ Object (readonly)

Returns the value of attribute supervisor.



9
10
11
 
# File 'lib/fractor/continuous_server.rb', line 9

def supervisor
  @supervisor
end

#work_queue ⇒ Object (readonly)

Returns the value of attribute work_queue.



9
10
11
 
# File 'lib/fractor/continuous_server.rb', line 9

def work_queue
  @work_queue
end

Instance Method Details

#on_error {|WorkResult| ... } ⇒ Object

Register a callback for errors

Yields:



37
38
39
 
# File 'lib/fractor/continuous_server.rb', line 37

def on_error(&block)
  @error_callbacks << block
end

#on_result {|WorkResult| ... } ⇒ Object

Register a callback for successful results

Yields:



31
32
33
 
# File 'lib/fractor/continuous_server.rb', line 31

def on_result(&block)
  @result_callbacks << block
end

#run ⇒ Object

Start the server and block until shutdown This method handles:

  • Opening log file if specified
  • Creating and starting supervisor
  • Registering result callbacks with ResultAggregator
  • Blocking until shutdown signal received


55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/fractor/continuous_server.rb', line 55

def run
  setup_log_file
  setup_supervisor
  register_result_callbacks
  start_supervisor_thread

  log_message("Continuous server started")
  log_message("Press Ctrl+C to stop")

  begin
    # Event-driven: simply join the supervisor thread
    # It will exit when @running = false and shutdown is complete
    @supervisor_thread&.join
  rescue Interrupt
    log_message("Interrupt received, shutting down...")
  ensure
    cleanup
  end
end

#start ⇒ Object

Start the server (alias for run). Provides a consistent API with stop method.

See Also:



45
46
47
 
# File 'lib/fractor/continuous_server.rb', line 45

def start
  run
end

#stop ⇒ Object

Stop the server programmatically



76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/fractor/continuous_server.rb', line 76

def stop
  return unless @running

  log_message("Stopping continuous server...")
  @running = false

  @supervisor&.stop

  # Ensure log file is closed
  # This is important when stop() is called from outside the run() thread
  # The run() method's ensure block will also call cleanup, but we ensure
  # it here as well for immediate cleanup
  cleanup
end