Class: Concurrent::Edge::Future

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

Overview

Represents a value which will become available in future. May fail with a reason instead.

Direct Known Subclasses

CompletableFuture

Instance Method Summary collapse

Methods inherited from Event

#chain, #completed?, #default_executor, #initialize, #inspect, #on_completion, #on_completion!, #pending?, #set, #to_s, #unscheduled?, #wait

Constructor Details

This class inherits a constructor from Concurrent::Edge::Event

Instance Method Details

#any(*futures) ⇒ Future Also known as: |

Returns which has first completed value from futures.

Returns:

  • (Future) —

    which has first completed value from futures



672
673
674
 
# File 'lib/concurrent/edge/future.rb', line 672

def any(*futures)
  AnyCompletePromise.new([self, *futures], @DefaultExecutor).future
end

#chain_completable(completable_future) ⇒ Object Also known as: tangle 



652
653
654
 
# File 'lib/concurrent/edge/future.rb', line 652

def chain_completable(completable_future)
  on_completion! { completable_future.complete_with internal_state }
end

#delay ⇒ Future

Inserts delay into the chain of Futures making rest of it lazy evaluated.

Returns:



678
679
680
 
# File 'lib/concurrent/edge/future.rb', line 678

def delay
  ZipFutureEventPromise.new(self, DelayPromise.new(@DefaultExecutor).future, @DefaultExecutor).future
end

#exception(*args) ⇒ Object

Examples:

allows failed Future to be risen

raise Concurrent.future.fail



629
630
631
632
633
634
635
636
637
638
 
# File 'lib/concurrent/edge/future.rb', line 629

def exception(*args)
  raise 'obligation is not failed' unless failed?
  reason = internal_state.reason
  if reason.is_a?(::Array)
    reason.each { |e| log ERROR, 'Edge::Future', e }
    Concurrent::Error.new 'multiple exceptions, inspect log'
  else
    reason.exception(*args)
  end
end

#failed?(state = internal_state) ⇒ Boolean

Has Future been failed?

Returns:

  • (Boolean) 


570
571
572
 
# File 'lib/concurrent/edge/future.rb', line 570

def failed?(state = internal_state)
  state.completed? && !state.success?
end

#flat(level = 1) ⇒ Object

zips with the Future in the value

Examples:

Concurrent.future { Concurrent.future { 1 } }.flat.value # => 1



667
668
669
 
# File 'lib/concurrent/edge/future.rb', line 667

def flat(level = 1)
  FlatPromise.new(self, level, @DefaultExecutor).future
end

#fulfilled? ⇒ Boolean

Returns:

  • (Boolean) 


563
564
565
566
 
# File 'lib/concurrent/edge/future.rb', line 563

def fulfilled?
  deprecated_method 'fulfilled?', 'success?'
  success?
end

#on_failure(executor = nil) {|reason| ... } ⇒ Object

Returns self.

Yields:

  • (reason) —

    executed async on executor when failed?

Returns:

  • self



732
733
734
 
# File 'lib/concurrent/edge/future.rb', line 732

def on_failure(executor = nil, &callback)
  add_callback :async_callback_on_failure, executor || @DefaultExecutor, callback
end

#on_failure! {|reason| ... } ⇒ Object

Returns self.

Yields:

  • (reason) —

    executed sync when failed?

Returns:

  • self



744
745
746
 
# File 'lib/concurrent/edge/future.rb', line 744

def on_failure!(&callback)
  add_callback :callback_on_failure, callback
end

#on_success(executor = nil) {|value| ... } ⇒ Object

Returns self.

Yields:

  • (value) —

    executed async on executor when success

Returns:

  • self



726
727
728
 
# File 'lib/concurrent/edge/future.rb', line 726

def on_success(executor = nil, &callback)
  add_callback :async_callback_on_success, executor || @DefaultExecutor, callback
end

#on_success! {|value| ... } ⇒ Object

Returns self.

Yields:

  • (value) —

    executed sync when success

Returns:

  • self



738
739
740
 
# File 'lib/concurrent/edge/future.rb', line 738

def on_success!(&callback)
  add_callback :callback_on_success, callback
end

#reason(timeout = nil) ⇒ Exception?

Note:

If the Future can have value nil then it cannot be distinquished from nil returned on timeout. In this case is better to use first wait then value (or similar).

Note:

a thread should wait only once! For repeated checking use faster completed? check. If thread waits periodically it will dangerously grow the waiters stack.

Returns the reason of the Future’s failure.

Returns:

  • (Exception, nil) —

    the reason of the Future’s failure



592
593
594
595
 
# File 'lib/concurrent/edge/future.rb', line 592

def reason(timeout = nil)
  touch
  internal_state.reason if wait_until_complete timeout
end

#rejected? ⇒ Boolean

Returns:

  • (Boolean) 


574
575
576
577
 
# File 'lib/concurrent/edge/future.rb', line 574

def rejected?
  deprecated_method 'rejected?', 'failed?'
  failed?
end

#rescue(executor = nil) {|reason| ... } ⇒ Future

Yields:

  • (reason) —

    executed only on parent failure

Returns:



660
661
662
 
# File 'lib/concurrent/edge/future.rb', line 660

def rescue(executor = nil, &callback)
  RescuePromise.new(self, @DefaultExecutor, executor || @DefaultExecutor, &callback).future
end

#result(timeout = nil) ⇒ Array(Boolean, Object, Exception)?

Note:

If the Future can have value nil then it cannot be distinquished from nil returned on timeout. In this case is better to use first wait then value (or similar).

Note:

a thread should wait only once! For repeated checking use faster completed? check. If thread waits periodically it will dangerously grow the waiters stack.

Returns triplet of success, value, reason.

Returns:

  • (Array(Boolean, Object, Exception), nil) —

    triplet of success, value, reason



600
601
602
603
 
# File 'lib/concurrent/edge/future.rb', line 600

def result(timeout = nil)
  touch
  internal_state.result if wait_until_complete timeout
end

#schedule(intended_time) ⇒ Future

Schedules rest of the chain for execution with specified time or on specified time

Returns:



684
685
686
687
688
689
690
 
# File 'lib/concurrent/edge/future.rb', line 684

def schedule(intended_time)
  chain do
    ZipFutureEventPromise.new(self,
                              ScheduledPromise.new(@DefaultExecutor, intended_time).event,
                              @DefaultExecutor).future
  end.flat
end

#state ⇒ :pending, ...

Returns:

  • (:pending, :success, :failed) 



# File 'lib/concurrent/edge/future.rb', line 554

#success?(state = internal_state) ⇒ Boolean

Has Future been success?

Returns:

  • (Boolean) 


559
560
561
 
# File 'lib/concurrent/edge/future.rb', line 559

def success?(state = internal_state)
  state.completed? && state.success?
end

#then(executor = nil) {|value| ... } ⇒ Future

Yields:

  • (value) —

    executed only on parent success

Returns:



642
643
644
 
# File 'lib/concurrent/edge/future.rb', line 642

def then(executor = nil, &callback)
  ThenPromise.new(self, @DefaultExecutor, executor || @DefaultExecutor, &callback).future
end

#then_ask(actor) ⇒ Future

Asks the actor with its value.

Returns:

  • (Future) —

    new future with the response form the actor



648
649
650
 
# File 'lib/concurrent/edge/future.rb', line 648

def then_ask(actor)
  self.then { |v| actor.ask(v) }.flat
end

#then_put(channel) ⇒ Object

Note:

may block

Note:

only proof of concept



720
721
722
 
# File 'lib/concurrent/edge/future.rb', line 720

def then_put(channel)
  on_success(:io) { |value| channel.put value }
end

#then_select(*channels) ⇒ Future

Zips with selected value form the suplied channels

Returns:



694
695
696
 
# File 'lib/concurrent/edge/future.rb', line 694

def then_select(*channels)
  ZipFuturesPromise.new([self, Concurrent.select(*channels)], @DefaultExecutor).future
end

#value(timeout = nil) ⇒ Object?

Note:

If the Future can have value nil then it cannot be distinquished from nil returned on timeout. In this case is better to use first wait then value (or similar).

Note:

a thread should wait only once! For repeated checking use faster completed? check. If thread waits periodically it will dangerously grow the waiters stack.

Returns the value of the Future when success, nil on timeout.

Returns:

  • (Object, nil) —

    the value of the Future when success, nil on timeout



584
585
586
587
 
# File 'lib/concurrent/edge/future.rb', line 584

def value(timeout = nil)
  touch
  internal_state.value if wait_until_complete timeout
end

#value!(timeout = nil) ⇒ Object?

Note:

If the Future can have value nil then it cannot be distinquished from nil returned on timeout. In this case is better to use first wait then value (or similar).

Note:

a thread should wait only once! For repeated checking use faster completed? check. If thread waits periodically it will dangerously grow the waiters stack.

Wait until Future is #complete?

Parameters:

  • timeout (Numeric) (defaults to: nil) —

    the maximum time in second to wait.

Returns:

  • (Object, nil)

Raises:

  • reason on failure



622
623
624
625
 
# File 'lib/concurrent/edge/future.rb', line 622

def value!(timeout = nil)
  touch
  internal_state.value if wait_until_complete! timeout
end

#wait!(timeout = nil) ⇒ Event, ...

Note:

a thread should wait only once! For repeated checking use faster completed? check. If thread waits periodically it will dangerously grow the waiters stack.

Wait until Future is #complete?

Parameters:

  • timeout (Numeric) (defaults to: nil) —

    the maximum time in second to wait.

Returns:

  • (Event, true, false) —

    self or true/false if timeout is used

Raises:

  • reason on failure



610
611
612
613
614
 
# File 'lib/concurrent/edge/future.rb', line 610

def wait!(timeout = nil)
  touch
  result = wait_until_complete!(timeout)
  timeout ? result : self
end

#with_default_executor(executor) ⇒ Future

Changes default executor for rest of the chain

Returns:



700
701
702
 
# File 'lib/concurrent/edge/future.rb', line 700

def with_default_executor(executor)
  FutureWrapperPromise.new(self, executor).future
end

#zip(other) ⇒ Future Also known as: &

Zip with future producing new Future

Returns:



706
707
708
709
710
711
712
 
# File 'lib/concurrent/edge/future.rb', line 706

def zip(other)
  if other.is_a?(Future)
    ZipFutureFuturePromise.new(self, other, @DefaultExecutor).future
  else
    ZipFutureEventPromise.new(self, other, @DefaultExecutor).future
  end
end