Module: RSMP::Task

Included in:
Node, Proxy
Defined in:
lib/rsmp/node/task.rb

Overview

Explicit ownership for a long-running Async task tree.

Defined Under Namespace

Classes: ConditionWaitTimeout

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#task ⇒ Object (readonly)

Returns the value of attribute task.



6
7
8
 
# File 'lib/rsmp/node/task.rb', line 6

def task
  @task
end

Instance Method Details

#initialize_task ⇒ Object 



8
9
10
11
 
# File 'lib/rsmp/node/task.rb', line 8

def initialize_task
  @task = nil
  @termination_completion = Completion.new
end

#restart ⇒ Object 



31
32
33
 
# File 'lib/rsmp/node/task.rb', line 31

def restart
  termination_completion.succeed(Termination.new(reason: :restart, source: self))
end

#run ⇒ Object 



43
44
45
 
# File 'lib/rsmp/node/task.rb', line 43

def run
  start_subtasks
end

#start(parent: Async::Task.current) ⇒ Object

Start under an explicit Async parent. A current task is accepted for convenience, but this library never creates a hidden root reactor.

Raises:

  • (ArgumentError) 


15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/rsmp/node/task.rb', line 15

def start(parent: Async::Task.current)
  return @task if @task&.running?

  valid_parent = parent.respond_to?(:async) && (!parent.respond_to?(:running?) || parent.running?)
  raise ArgumentError, 'an active Async parent task is required' unless valid_parent

  child = parent.async do |task|
    task.annotate "#{self.class.name} main task"
    @task = task
    run
  ensure
    stop_subtasks
  end
  @task = child
end

#stop ⇒ Object 



53
54
55
 
# File 'lib/rsmp/node/task.rb', line 53

def stop
  stop_task
end

#stop_subtasks ⇒ Object 



57
 
# File 'lib/rsmp/node/task.rb', line 57

def stop_subtasks; end

#stop_task ⇒ Object 



59
60
61
62
63
64
65
 
# File 'lib/rsmp/node/task.rb', line 59

def stop_task
  task = @task
  return unless task&.running?

  task.cancel
  task.wait unless task.current?
end

#task_status ⇒ Object 



39
40
41
 
# File 'lib/rsmp/node/task.rb', line 39

def task_status
  @task&.status
end

#wait ⇒ Object

Async::Task#wait raises unexpected child failures and returns nil for cancellation, matching Async's native task contract.



49
50
51
 
# File 'lib/rsmp/node/task.rb', line 49

def wait
  @task&.wait
end

#wait_for_condition(condition, timeout:, task: Async::Task.current, &block) ⇒ Object

Wait for an edge-triggered condition and return an expected timeout as a Result. Async cancellation is not rescued and therefore propagates.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/rsmp/node/task.rb', line 69

def wait_for_condition(condition, timeout:, task: Async::Task.current, &block)
  raise ArgumentError, 'an active Async task is required' unless task&.running?

  value = task.with_timeout(timeout, ConditionWaitTimeout) do
    loop do
      signalled = condition.wait
      break signalled unless block

      matched = yield(signalled)
      break matched if matched
    end
  end
  Result.success(value)
rescue ConditionWaitTimeout => e
  Result.failure(
    :timeout,
    message: "Condition was not met within #{timeout}s",
    source: :timeout,
    cause: e
  )
end

#wait_for_condition! ⇒ Object 



91
92
93
 
# File 'lib/rsmp/node/task.rb', line 91

def wait_for_condition!(...)
  wait_for_condition(...).value!
end

#wait_for_termination ⇒ Object 



35
36
37
 
# File 'lib/rsmp/node/task.rb', line 35

def wait_for_termination
  termination_completion.wait
end