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:



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

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



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

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



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

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:



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

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:



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

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



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

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:



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

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) 


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

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

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

post job on :fast executor

Returns:

  • (true, false) 


103
104
105
 
# File 'lib/concurrent/edge/future.rb', line 103

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

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

post job on executor

Returns:

  • (true, false) 


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

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:



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

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:



95
96
97
98
99
 
# File 'lib/concurrent/edge/future.rb', line 95

def select(*channels)
  probe = future
  channels.each { |ch| ch.select probe }
  probe
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



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

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:



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

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