Module: Concurrent::Edge::FutureShortcuts

Included in:
Concurrent, Concurrent, Concurrent::Edge, Concurrent::Edge
Defined in:
lib/concurrent/edge/future.rb

Overview

Provides edge features, which will be added to or replace features in main gem.

Contains new unified implementation of Futures and Promises which combines Features of previous ‘Future`, `Promise`, `IVar`, `Event`, `Probe`, `dataflow`, `Delay`, `TimerTask` into single framework. It uses extensively new synchronization layer to make all the paths lock-free with exception of blocking threads on `#wait`. It offers better performance and does not block threads (exception being #wait and similar methods where it’s intended).

## Examples

Instance Method Summary collapse

Instance Method Details

#any(*futures) ⇒ Future

Constructs new Concurrent::Edge::Future which is completed after first of the futures is complete.

Parameters:

Returns:



90
91
92
 
# File 'lib/concurrent/edge/future.rb', line 90

def any(*futures)
  AnyPromise.new(futures, :io).future
end

#completed_event(default_executor = :io) ⇒ Event

Returns which is already completed.

Returns:

  • (Event) —

    which is already completed



59
60
61
 
# File 'lib/concurrent/edge/future.rb', line 59

def completed_event(default_executor = :io)
  ImmediateEventPromise.new(default_executor).event
end

#completed_future(success, value, reason, default_executor = :io) ⇒ Future

Returns which is already completed.

Returns:

  • (Future) —

    which is already completed



44
45
46
 
# File 'lib/concurrent/edge/future.rb', line 44

def completed_future(success, value, reason, default_executor = :io)
  ImmediateFuturePromise.new(default_executor, success, value, reason).future
end

#delay(default_executor = :io, &task) ⇒ Future

Constructs new Future which will evaluate to the block after requested by calling ‘#wait`, `#value`, `#value!`, etc. on it or on any of the chained futures.

Returns:



68
69
70
 
# File 'lib/concurrent/edge/future.rb', line 68

def delay(default_executor = :io, &task)
  Delay.new(default_executor).future.then(&task)
end

#event(default_executor = :io) ⇒ CompletableEvent

User is responsible for completing the event once by CompletableEvent#complete

Returns:



25
26
27
 
# File 'lib/concurrent/edge/future.rb', line 25

def event(default_executor = :io)
  CompletableEventPromise.new(default_executor).future
end

#failed_future(reason, default_executor = :io) ⇒ Future

Returns which is already completed in failed state with reason.

Returns:

  • (Future) —

    which is already completed in failed state with reason



54
55
56
 
# File 'lib/concurrent/edge/future.rb', line 54

def failed_future(reason, default_executor = :io)
  completed_future false, nil, reason, default_executor
end

#future(default_executor = :io, &task) ⇒ Future #future(default_executor = :io) ⇒ CompletableFuture Also known as: async

Overloads:



35
36
37
38
39
40
41
 
# File 'lib/concurrent/edge/future.rb', line 35

def future(default_executor = :io, &task)
  if task
    ImmediateEventPromise.new(default_executor).future.then(&task)
  else
    CompletableFuturePromise.new(default_executor).future
  end
end

#post(*args, &job) ⇒ true, false

post job on :io executor

Returns:

  • (true, false) 


114
115
116
 
# File 'lib/concurrent/edge/future.rb', line 114

def post(*args, &job)
  post_on(:io, *args, &job)
end

#post!(*args, &job) ⇒ true, false

post job on :fast executor

Returns:

  • (true, false) 


108
109
110
 
# File 'lib/concurrent/edge/future.rb', line 108

def post!(*args, &job)
  post_on(:fast, *args, &job)
end

#post_on(executor, *args, &job) ⇒ true, false

post job on executor

Returns:

  • (true, false) 


120
121
122
 
# File 'lib/concurrent/edge/future.rb', line 120

def post_on(executor, *args, &job)
  Concurrent.executor(executor).post(*args, &job)
end

#schedule(intended_time, default_executor = :io, &task) ⇒ Future

Schedules the block to be executed on executor in given intended_time.

Parameters:

  • intended_time (Numeric, Time) —

    Numeric => run in ‘intended_time` seconds. Time => eun on time.

Returns:



75
76
77
 
# File 'lib/concurrent/edge/future.rb', line 75

def schedule(intended_time, default_executor = :io, &task)
  ScheduledPromise.new(default_executor, intended_time).future.then(&task)
end

#select(*channels) ⇒ Future

only proof of concept

Returns:



96
97
98
99
100
101
102
103
104
 
# File 'lib/concurrent/edge/future.rb', line 96

def select(*channels)
  future do
    Channel.select do |s|
      channels.each do |ch|
        s.take(ch) { |value| [value, ch] }
      end
    end
  end
end

#succeeded_future(value, default_executor = :io) ⇒ Future

Returns which is already completed in success state with value.

Returns:

  • (Future) —

    which is already completed in success state with value



49
50
51
 
# File 'lib/concurrent/edge/future.rb', line 49

def succeeded_future(value, default_executor = :io)
  completed_future true, value, nil, default_executor
end

#zip(*futures) ⇒ Future

Constructs new Concurrent::Edge::Future which is completed after all futures are complete. Its value is array of dependent future values. If there is an error it fails with the first one.

Parameters:

Returns:



83
84
85
 
# File 'lib/concurrent/edge/future.rb', line 83

def zip(*futures)
  ZipPromise.new(futures, :io).future
end