Class: Rx::Concurrent::Future

Inherits:
Object
  • Object
show all
Defined in:
lib/rx/concurrent/future.rb

Constant Summary collapse

ALLOWED_STATES = 
%i[pending in_progress completed failed]
@@pool = 
ThreadPool.new.start

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(&block) ⇒ Future

Returns a new instance of Future.



21
22
23
24
25
 
# File 'lib/rx/concurrent/future.rb', line 21

def initialize(&block)
  @channel = Queue.new
  @state = :pending
  @work = block
end

Instance Attribute Details

#error ⇒ Object (readonly)

Returns the value of attribute error.



11
12
13
 
# File 'lib/rx/concurrent/future.rb', line 11

def error
  @error
end

Class Method Details

.execute(&block) ⇒ Object 



13
14
15
 
# File 'lib/rx/concurrent/future.rb', line 13

def self.execute(&block)
  Future.new(&block).execute
end

.thread_pool ⇒ Object 



17
18
19
 
# File 'lib/rx/concurrent/future.rb', line 17

def self.thread_pool
  @@pool
end

Instance Method Details

#completed? ⇒ Boolean

Returns:

  • (Boolean) 


27
28
29
 
# File 'lib/rx/concurrent/future.rb', line 27

def completed?
  state == :completed
end

#execute ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/rx/concurrent/future.rb', line 31

def execute
  @state = :in_progress
  pool.submit do
    begin
      channel << execute_work(&work)
      @state = :completed
    rescue StandardError => ex
      @error = ex
      @state = :failed
      channel.close
    end
  end

  self
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


47
48
49
 
# File 'lib/rx/concurrent/future.rb', line 47

def failed?
  state == :failed
end

#in_progress? ⇒ Boolean

Returns:

  • (Boolean) 


51
52
53
 
# File 'lib/rx/concurrent/future.rb', line 51

def in_progress?
  state == :in_progress
end

#pending? ⇒ Boolean

Returns:

  • (Boolean) 


55
56
57
 
# File 'lib/rx/concurrent/future.rb', line 55

def pending?
  state == :pending
end

#value ⇒ Object 



59
60
61
62
63
64
65
 
# File 'lib/rx/concurrent/future.rb', line 59

def value
  if (completed? || failed?) && channel.empty?
    return @value
  end

  @value = channel.pop
end