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



156
157
158
159
 
# File 'lib/omq/routing/fair_queue.rb', line 156

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) 


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

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

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


183
184
185
 
# File 'lib/omq/routing/fair_queue.rb', line 183

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


167
168
169
170
 
# File 'lib/omq/routing/fair_queue.rb', line 167

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

#push(item) ⇒ void

This method returns an undefined value.

Parameters:

  • item (Object, nil) 


191
192
193
 
# File 'lib/omq/routing/fair_queue.rb', line 191

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