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.

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



12
13
14
 
# File 'lib/farce/abstract/timer_queue.rb', line 12

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

Instance Method Details

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

Delete the oldest matching value at the exact time.

Parameters:

  • value (BasicObject) —

    the value to delete

  • at (Numeric, Time) —

    the exact time to search

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

    compare values by identity instead of equality

Returns:

  • (Boolean) —

    whether a value was deleted



108
109
110
 
# File 'lib/farce/abstract/timer_queue.rb', line 108

def delete(value, at:, compare_by_identity: false)
  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



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

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



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

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) 


82
83
84
85
 
# File 'lib/farce/abstract/timer_queue.rb', line 82

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



90
91
92
93
94
 
# File 'lib/farce/abstract/timer_queue.rb', line 90

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



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

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



35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/farce/abstract/timer_queue.rb', line 35

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) ⇒ Boolean

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:

  • (Boolean) —

    whether the value was added

Raises:

  • (ThreadError) —

    when the queue is at capacity and non_block is true



26
27
28
29
 
# File 'lib/farce/abstract/timer_queue.rb', line 26

def push(value, non_block = false, timeout: nil, **time_options) # rubocop:disable Style/OptionalBooleanParameter
  at = Clock.parse(time_options)
  push_to_storage(at, non_block, value, timeout:)
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



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

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

#try_push(value, **time_options) { ... } ⇒ Boolean, 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:

  • (Boolean, BasicObject) —

    true, or the fallback result when full



54
55
56
57
58
 
# File 'lib/farce/abstract/timer_queue.rb', line 54

def try_push(value, **time_options)
  at = Clock.parse(time_options)
  return true 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



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

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