Class: OMQ::Routing::SignalingQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/routing/fair_queue.rb

Overview

Wraps a per-connection bounded queue so that each #enqueue also signals the FairQueue to wake a blocked #dequeue.

Instance Method Summary collapse

Constructor Details

#initialize(inner, fair_queue) ⇒ SignalingQueue

Returns a new instance of SignalingQueue.

Parameters:

  • inner (Async::LimitedQueue) —

    the per-connection bounded queue

  • fair_queue (FairQueue) —

    the parent fair queue to signal on enqueue



150
151
152
153
 
# File 'lib/omq/routing/fair_queue.rb', line 150

def initialize(inner, fair_queue)
  @inner = inner
  @fair  = fair_queue
end

Instance Method Details

#dequeue(timeout: nil) ⇒ Array<String>?

Parameters:

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

    dequeue timeout

Returns:

  • (Array<String>, nil) 


170
171
172
 
# File 'lib/omq/routing/fair_queue.rb', line 170

def dequeue(timeout: nil)
  @inner.dequeue(timeout: timeout)
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


177
178
179
 
# File 'lib/omq/routing/fair_queue.rb', line 177

def empty?
  @inner.empty?
end

#enqueue(msg) ⇒ void

This method returns an undefined value.

Enqueues a message and signals the fair queue.

Parameters:

  • msg (Array<String>) 


161
162
163
164
 
# File 'lib/omq/routing/fair_queue.rb', line 161

def enqueue(msg)
  @inner.enqueue(msg)
  @fair.signal
end

#push(item) ⇒ void

This method returns an undefined value.

Parameters:

  • item (Object, nil) 


185
186
187
 
# File 'lib/omq/routing/fair_queue.rb', line 185

def push(item)
  @inner.push(item)
end