Class: Taski::Execution::Registry

Inherits:
Object
  • Object
show all
Defined in:
lib/taski/execution/registry.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Registry

Returns a new instance of Registry.



8
9
10
11
12
13
 
# File 'lib/taski/execution/registry.rb', line 8

def initialize
  @tasks = {}
  @threads = []
  @monitor = Monitor.new
  @abort_requested = false
end

Instance Method Details

#abort_requested? ⇒ Boolean

Returns true if abort has been requested.

Returns:

  • (Boolean) —

    true if abort has been requested



60
61
62
 
# File 'lib/taski/execution/registry.rb', line 60

def abort_requested?
  @monitor.synchronize { @abort_requested }
end

#create_wrapper(task_class, execution_facade:) ⇒ TaskWrapper

Create or retrieve a TaskWrapper for the given task class. Encapsulates the standard wrapper creation pattern used by Executor and WorkerPool.

Parameters:

  • task_class (Class) —

    The task class

  • execution_facade (ExecutionFacade) —

    The execution facade

Returns:



83
84
85
86
87
88
89
 
# File 'lib/taski/execution/registry.rb', line 83

def create_wrapper(task_class, execution_facade:)
  get_or_create(task_class) do
    task_instance = task_class.allocate
    task_instance.send(:initialize)
    TaskWrapper.new(task_instance, registry: self, execution_facade: execution_facade)
  end
end

#failed_clean_wrappers ⇒ Array<TaskWrapper>

Returns All wrappers that have clean errors.

Returns:

  • (Array<TaskWrapper>) —

    All wrappers that have clean errors



72
73
74
75
76
 
# File 'lib/taski/execution/registry.rb', line 72

def failed_clean_wrappers
  @monitor.synchronize do
    @tasks.values.select { |w| w.clean_error }
  end
end

#failed_wrappers ⇒ Array<TaskWrapper>

Returns All wrappers that have errors.

Returns:

  • (Array<TaskWrapper>) —

    All wrappers that have errors



65
66
67
68
69
 
# File 'lib/taski/execution/registry.rb', line 65

def failed_wrappers
  @monitor.synchronize do
    @tasks.values.select { |w| w.error }
  end
end

#get_or_create(task_class) { ... } ⇒ Object

Returns The task instance.

Parameters:

  • task_class (Class) —

    The task class

Yields:

  • Block to create the task instance if it doesn't exist

Returns:

  • (Object) —

    The task instance



18
19
20
21
22
 
# File 'lib/taski/execution/registry.rb', line 18

def get_or_create(task_class)
  @monitor.synchronize do
    @tasks[task_class] ||= yield
  end
end

#register(task_class, wrapper) ⇒ Object

Parameters:

  • task_class (Class) —

    The task class

  • wrapper (TaskWrapper) —

    The wrapper instance to register



26
27
28
 
# File 'lib/taski/execution/registry.rb', line 26

def register(task_class, wrapper)
  @monitor.synchronize { @tasks[task_class] = wrapper }
end

#register_thread(thread) ⇒ Object

Parameters:

  • thread (Thread) —

    The thread to register



38
39
40
 
# File 'lib/taski/execution/registry.rb', line 38

def register_thread(thread)
  @monitor.synchronize { @threads << thread }
end

#registered?(task_class) ⇒ Boolean

Check if a task wrapper has been registered (created during run phase).

Parameters:

  • task_class (Class) —

    The task class

Returns:

  • (Boolean) —

    true if a wrapper exists for this task



33
34
35
 
# File 'lib/taski/execution/registry.rb', line 33

def registered?(task_class)
  @monitor.synchronize { @tasks.key?(task_class) }
end

#request_abort! ⇒ Object 



55
56
57
 
# File 'lib/taski/execution/registry.rb', line 55

def request_abort!
  @monitor.synchronize { @abort_requested = true }
end

#reset! ⇒ Object 



47
48
49
50
51
52
53
 
# File 'lib/taski/execution/registry.rb', line 47

def reset!
  @monitor.synchronize do
    @tasks.clear
    @threads.clear
    @abort_requested = false
  end
end

#wait_all ⇒ Object 



42
43
44
45
 
# File 'lib/taski/execution/registry.rb', line 42

def wait_all
  threads = @monitor.synchronize { @threads.dup }
  threads.each(&:join)
end