Class: PgPipeline::Runtime::Task

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_pipeline/runtime/task.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name: nil) ⇒ Task

Returns a new instance of Task.



16
17
18
19
20
21
22
 
# File 'lib/pg_pipeline/runtime/task.rb', line 16

def initialize(name: nil)
  @fiber, @scheduler, @blocker, @result, @error = nil, nil, nil, nil, nil
  @done, @cancelled = false, false

  @name = name
  @waiters = []
end

Instance Attribute Details

#error ⇒ Object (readonly)

Returns the value of attribute error.



6
7
8
 
# File 'lib/pg_pipeline/runtime/task.rb', line 6

def error
  @error
end

#fiber ⇒ Object (readonly)

Returns the value of attribute fiber.



6
7
8
 
# File 'lib/pg_pipeline/runtime/task.rb', line 6

def fiber
  @fiber
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



6
7
8
 
# File 'lib/pg_pipeline/runtime/task.rb', line 6

def name
  @name
end

Class Method Details

.spawn(name: nil, &block) ⇒ Object

Raises:

  • (ArgumentError) 


8
9
10
11
12
13
14
 
# File 'lib/pg_pipeline/runtime/task.rb', line 8

def self.spawn(name: nil, &block)
  raise ArgumentError, "block required" unless block

  task = new(name: name)
  task.__send__(:start, &block)
  task
end

Instance Method Details

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


42
 
# File 'lib/pg_pipeline/runtime/task.rb', line 42

def cancelled? = @cancelled

#enter_block(waiter) ⇒ Object 



48
49
50
 
# File 'lib/pg_pipeline/runtime/task.rb', line 48

def enter_block(waiter)
  @blocker = waiter
end

#exit_block ⇒ Object 



52
53
54
 
# File 'lib/pg_pipeline/runtime/task.rb', line 52

def exit_block
  @blocker = nil
end

#finished? ⇒ Boolean

Returns:

  • (Boolean) 


56
 
# File 'lib/pg_pipeline/runtime/task.rb', line 56

def finished? = @done

#raise_if_cancelled! ⇒ Object

Raises:



44
45
46
 
# File 'lib/pg_pipeline/runtime/task.rb', line 44

def raise_if_cancelled!
  raise Cancel, "task stopped" if @cancelled
end

#stop ⇒ Object 



31
32
33
34
35
36
37
38
39
40
 
# File 'lib/pg_pipeline/runtime/task.rb', line 31

def stop
  return false if @done

  @cancelled = true

  return true if interrupt_fiber
  return true if release_blocker

  false
end

#wait(timeout = nil) ⇒ Object

Raises:



24
25
26
27
28
29
 
# File 'lib/pg_pipeline/runtime/task.rb', line 24

def wait(timeout = nil)
  Runtime.with_timeout(timeout) { join }
  raise @error if @error

  @result
end