Class: Concurrent::Edge::Event

Inherits:
Synchronization::LockableObject
  • 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.



177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/concurrent/edge/future.rb', line 177

def initialize(promise, default_executor)
  super()
  @Promise            = promise
  @DefaultExecutor    = default_executor
  @Touched            = AtomicBoolean.new(false)
  @Callbacks          = LockFreeStack.new
  # TODO (pitr 12-Sep-2015): replace with AtomicFixnum, avoid aba problem
  # TODO (pitr 12-Sep-2015): look at java.util.concurrent solution
  @Waiters            = LockFreeStack.new
  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



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

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

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



246
247
248
 
# File 'lib/concurrent/edge/future.rb', line 246

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) 


208
209
210
 
# File 'lib/concurrent/edge/future.rb', line 208

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

#default_executor ⇒ Executor

Returns current default executor.

Returns:

  • (Executor) —

    current default executor

See Also:



235
236
237
 
# File 'lib/concurrent/edge/future.rb', line 235

def default_executor
  @DefaultExecutor
end

#delay ⇒ Event

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

Returns:



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

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

#inspect ⇒ Object 



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

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



288
289
290
 
# File 'lib/concurrent/edge/future.rb', line 288

def on_completion(executor = nil, &callback)
  add_callback :pr_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



294
295
296
 
# File 'lib/concurrent/edge/future.rb', line 294

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

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

Is Event/Future pending?

Returns:

  • (Boolean) 


196
197
198
 
# File 'lib/concurrent/edge/future.rb', line 196

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

#set(*args, &block) ⇒ Object 



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

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) 


190
191
192
 
# File 'lib/concurrent/edge/future.rb', line 190

def state
  internal_state.to_sym
end

#then_select(*channels) ⇒ Future

Zips with selected value form the suplied channels

Returns:



282
283
284
 
# File 'lib/concurrent/edge/future.rb', line 282

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

#to_s ⇒ Object 



304
305
306
 
# File 'lib/concurrent/edge/future.rb', line 304

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

#unscheduled? ⇒ Boolean

Returns:

  • (Boolean) 


200
201
202
 
# File 'lib/concurrent/edge/future.rb', line 200

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



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

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:



300
301
302
 
# File 'lib/concurrent/edge/future.rb', line 300

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

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

Zip with future producing new Future

Returns:



254
255
256
257
258
259
260
 
# File 'lib/concurrent/edge/future.rb', line 254

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