Class: Farce::Abstract::TimerQueue Abstract

Inherits:
Queue
  • Object
show all
Includes:
Internal::BlockingPriorityQueue
Defined in:
lib/farce/abstract/timer_queue.rb

Overview

This class is abstract.

Shared timer scheduling and blocking behavior.

Defined Under Namespace

Classes: Entry

Instance Attribute Summary

Attributes inherited from Queue

#capacity, #mode, #num_waiting, #size

Instance Method Summary collapse

Methods inherited from Queue

#age_tracking?, #clear, #close, #closed?, #deq, #duplicable?, #empty?, #enq, #full?, #generation, #length, #max, #oldest_age, #oldest_enqueued_at, #seal, #sealed?, #wait_push

Constructor Details

#initialize(capacity: nil, track_age: false) ⇒ TimerQueue

Returns a new instance of TimerQueue.

Parameters:

  • capacity (Integer, nil) (defaults to: nil) —

    the maximum number of values, or nil for an unbounded queue



43
44
45
 
# File 'lib/farce/abstract/timer_queue.rb', line 43

def initialize(capacity: nil, track_age: false)
  super
end

Instance Method Details

#delete(value, at: UNDEFINED, compare_by_identity: false) ⇒ Boolean

Cancel an entry, or delete the oldest matching value at the exact time. Pass at: to delete an Entry that was itself queued as a value.

Parameters:

  • value (Entry, BasicObject) —

    the cancellation handle or value to delete

  • at (Numeric, Time) (defaults to: UNDEFINED) —

    the exact time to search, required when deleting a value

  • compare_by_identity (Boolean) (defaults to: false) —

    compare values by identity instead of equality when at: is given

Returns:

  • (Boolean) —

    whether a value was deleted, or false if the entry belongs to another queue

Raises:

  • (ArgumentError) —

    when deleting a value without at:



143
144
145
146
147
148
149
 
# File 'lib/farce/abstract/timer_queue.rb', line 143

def delete(value, at: UNDEFINED, compare_by_identity: false)
  if UNDEFINED.equal?(at)
    raise ArgumentError, "at: is required when deleting a value" unless Entry === value
    return value.queue.equal?(self) && value.delete
  end
  delete_from_storage(Clock.at(at), value, compare_by_identity:)
end

#first_timestamp { ... } ⇒ Float?

Return the earliest timestamp without removing it.

Yields:

  • called when the queue is empty

Returns:

  • (Float, nil) —

    the timestamp or the fallback result



106
 
# File 'lib/farce/abstract/timer_queue.rb', line 106

def first_timestamp(&) = internal_queue.peek_priority(&)

#last_timestamp { ... } ⇒ Float?

Return the latest timestamp without removing it.

Yields:

  • called when the queue is empty

Returns:

  • (Float, nil) —

    the timestamp or the fallback result



111
 
# File 'lib/farce/abstract/timer_queue.rb', line 111

def last_timestamp(&) = internal_queue.peek_last_priority(&)

#overdue?(leeway: 0) ⇒ Boolean

Checks whether the earliest timestamp has been reached.

Parameters:

  • leeway (Numeric) (defaults to: 0) —

    how much leeway to allow for clock drift and scheduling delays

Returns:

  • (Boolean) 


115
116
117
118
 
# File 'lib/farce/abstract/timer_queue.rb', line 115

def overdue?(leeway: 0)
  return false unless timestamp = first_timestamp
  timestamp <= Clock.in(leeway)
end

#overdue_by(leeway: 0) ⇒ Float?

Returns how long the earliest timestamp has been overdue, or nil if it is not yet due.

Parameters:

  • leeway (Numeric) (defaults to: 0) —

    how much leeway to allow for clock drift and scheduling delays

Returns:

  • (Float, nil) —

    the number of seconds overdue, or nil if the earliest timestamp is not yet due



123
124
125
126
127
 
# File 'lib/farce/abstract/timer_queue.rb', line 123

def overdue_by(leeway: 0)
  return unless timestamp = first_timestamp
  delay = Clock.now - timestamp
  delay > leeway ? delay : nil
end

#peek { ... } ⇒ BasicObject?

Return the earliest value without removing it, whether or not it is due.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



101
 
# File 'lib/farce/abstract/timer_queue.rb', line 101

def peek(&) = internal_queue.peek(&)

#pop(non_block = false, timeout: nil) { ... } ⇒ BasicObject?

Remove the earliest value, waiting until its timestamp is reached.

Parameters:

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

    maximum number of seconds to wait

Yields:

  • called when the timeout expires first

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/farce/abstract/timer_queue.rb', line 67

def pop(non_block = false, timeout: nil) # rubocop:disable Style/OptionalBooleanParameter
  return try_pop { raise ThreadError, "queue empty" } if non_block
  deadline = timeout_at(timeout) unless timeout.nil?

  while true
    empty = false
    value = internal_queue.pop_before(Clock.now) { empty = true }
    return value unless empty
    return block_given? ? yield : nil if UNDEFINED.equal?(wait_for_timestamp(deadline))
  end
end

#push(value, non_block = false, timeout: nil, **time_options) ⇒ Entry, false

Add a value to become available at the given time.

Parameters:

  • value (BasicObject) —

    the value to add

  • non_block (Boolean) (defaults to: false) —

    whether to raise an exception when the queue is at capacity

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

    maximum number of seconds to wait for capacity

  • time_options (Hash{Symbol => Object}) —

    a scheduling option accepted by Clock.parse: at:, time:, timeout_at:, delay:, in:, offset:, wait:, or clock:. With no scheduling option, the value is available immediately. Since timeout: controls the capacity wait, use delay: or wait: to schedule a relative offset.

Returns:

  • (Entry, false) —

    the cancellation handle, or false if the capacity wait expires

Raises:

  • (ThreadError) —

    when the queue is at capacity and non_block is true



57
58
59
60
61
 
# File 'lib/farce/abstract/timer_queue.rb', line 57

def push(value, non_block = false, timeout: nil, **time_options) # rubocop:disable Style/OptionalBooleanParameter
  at = Clock.parse(time_options)
  entry = Entry.new(value, at, self)
  push_to_storage(at, non_block, value, timeout:) && entry
end

#try_pop { ... } ⇒ BasicObject?

Try to remove the earliest value only if its timestamp has been reached.

Yields:

  • called when no value is ready

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



96
 
# File 'lib/farce/abstract/timer_queue.rb', line 96

def try_pop(&) = internal_queue.pop_before(Clock.now, &)

#try_push(value, **time_options) { ... } ⇒ Entry, BasicObject

Try to add a value without waiting for capacity.

Parameters:

  • value (BasicObject) —

    the value to add

  • time_options (Hash{Symbol => Object}) —

    a scheduling option accepted by Clock.parse: at:, time:, timeout_at:, delay:, in:, offset:, timeout:, wait:, or clock:. With no scheduling option, the value is available immediately.

Yields:

  • called when the queue is at capacity

Returns:

  • (Entry, BasicObject) —

    the cancellation handle, or the fallback result when full



86
87
88
89
90
91
 
# File 'lib/farce/abstract/timer_queue.rb', line 86

def try_push(value, **time_options)
  at = Clock.parse(time_options)
  entry = Entry.new(value, at, self)
  return entry if internal_queue.push(at, value)
  block_given? ? yield : false
end

#wait_pop(timeout: nil) ⇒ Boolean

Wait until the earliest timestamp is reached without removing its value.

Parameters:

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

    maximum number of seconds to wait

Returns:

  • (Boolean) —

    whether a value is ready



132
133
134
 
# File 'lib/farce/abstract/timer_queue.rb', line 132

def wait_pop(timeout: nil) # rubocop:disable Naming/PredicateMethod
  !UNDEFINED.equal?(wait_for_timestamp(timeout_at(timeout)))
end