Class: Concurrent::Edge::Event

Inherits:
Synchronization::Object
  • Object
show all
Includes:
Concern::Deprecation, Concern::Logging
Defined in:
lib/concurrent/edge/future.rb

Overview

Represents an event which will happen in future (will be completed). It has to always happen.

Direct Known Subclasses

CompletableEvent, Future

Instance Method Summary collapse

Constructor Details

#initialize(promise, default_executor) ⇒ Event

Returns a new instance of Event.



197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/concurrent/edge/future.rb', line 197

def initialize(promise, default_executor)
  super()
  @Lock               = Mutex.new
  @Condition          = ConditionVariable.new
  @Promise            = promise
  @DefaultExecutor    = default_executor
  @Touched            = AtomicBoolean.new false
  @Callbacks          = LockFreeStack.new
  @Waiters            = AtomicFixnum.new 0
  self.internal_state = PENDING
end

Instance Method Details

#chain(executor = nil) {|success, value, reason| ... } ⇒ Object Also known as: then

Yields:

  • (success, value, reason) —

    of the parent



260
261
262
 
# File 'lib/concurrent/edge/future.rb', line 260

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

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



266
267
268
 
# File 'lib/concurrent/edge/future.rb', line 266

def chain_completable(completable_event)
  on_completion! { completable_event.complete_with COMPLETED }
end

#completed?(state = internal_state) ⇒ Boolean Also known as: complete?

Has the Event been completed?

Returns:

  • (Boolean) 


228
229
230
 
# File 'lib/concurrent/edge/future.rb', line 228

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

#default_executor ⇒ Executor

Returns current default executor.

Returns:

  • (Executor) —

    current default executor

See Also:



255
256
257
 
# File 'lib/concurrent/edge/future.rb', line 255

def default_executor
  @DefaultExecutor
end

#delay ⇒ Event

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

Returns:



286
287
288
 
# File 'lib/concurrent/edge/future.rb', line 286

def delay
  ZipEventEventPromise.new(self, DelayPromise.new(@DefaultExecutor).event, @DefaultExecutor).event
end

#inspect ⇒ Object 



328
329
330
 
# File 'lib/concurrent/edge/future.rb', line 328

def inspect
  "#{to_s[0..-2]} blocks:[#{blocks.map(&:to_s).join(', ')}]>"
end

#on_completion(executor = nil) {|success, value, reason| ... } ⇒ Object

Returns self.

Yields:

  • (success, value, reason) —

    executed async on ‘executor` when completed

Returns:

  • self



308
309
310
 
# File 'lib/concurrent/edge/future.rb', line 308

def on_completion(executor = nil, &callback)
  add_callback :async_callback_on_completion, executor || @DefaultExecutor, callback
end

#on_completion! {|success, value, reason| ... } ⇒ Object

Returns self.

Yields:

  • (success, value, reason) —

    executed sync when completed

Returns:

  • self



314
315
316
 
# File 'lib/concurrent/edge/future.rb', line 314

def on_completion!(&callback)
  add_callback :callback_on_completion, callback
end

#pending?(state = internal_state) ⇒ Boolean Also known as: incomplete?

Is Event/Future pending?

Returns:

  • (Boolean) 


216
217
218
 
# File 'lib/concurrent/edge/future.rb', line 216

def pending?(state = internal_state)
  !state.completed?
end

#set(*args, &block) ⇒ Object 



332
333
334
335
 
# File 'lib/concurrent/edge/future.rb', line 332

def set(*args, &block)
  raise 'Use CompletableEvent#complete or CompletableFuture#complete instead, ' +
            'constructed by Concurrent.event or Concurrent.future respectively.'
end

#state ⇒ :pending, :completed

Returns:

  • (:pending, :completed) 


210
211
212
 
# File 'lib/concurrent/edge/future.rb', line 210

def state
  internal_state.to_sym
end

#then_select(*channels) ⇒ Future

Zips with selected value form the suplied channels

Returns:



302
303
304
 
# File 'lib/concurrent/edge/future.rb', line 302

def then_select(*channels)
  ZipFutureEventPromise(Concurrent.select(*channels), self, @DefaultExecutor).future
end

#to_s ⇒ Object 



324
325
326
 
# File 'lib/concurrent/edge/future.rb', line 324

def to_s
  "<##{self.class}:0x#{'%x' % (object_id << 1)} #{state.to_sym}>"
end

#unscheduled? ⇒ Boolean

Returns:

  • (Boolean) 


220
221
222
 
# File 'lib/concurrent/edge/future.rb', line 220

def unscheduled?
  raise 'unsupported'
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 Event 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



240
241
242
243
244
 
# File 'lib/concurrent/edge/future.rb', line 240

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

#with_default_executor(executor) ⇒ Event

Changes default executor for rest of the chain

Returns:



320
321
322
 
# File 'lib/concurrent/edge/future.rb', line 320

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

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

Zip with future producing new Future

Returns:



274
275
276
277
278
279
280
 
# File 'lib/concurrent/edge/future.rb', line 274

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