Class: Asynchronic::QueueEngine::Ost

Inherits:
Object
  • Object
show all
Defined in:
lib/asynchronic/queue_engine/ost.rb

Defined Under Namespace

Classes: Listener, Queue

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Ost

Returns a new instance of Ost.



7
8
9
10
11
12
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 7

def initialize(options={})
  @redis = Asynchronic.establish_redis_connection options
  @default_queue = options.fetch(:default_queue, Asynchronic.default_queue)
  @queues ||= Hash.new { |h,k| h[k] = Queue.new k, redis }
  @keep_alive_thread = notify_keep_alive
end

Instance Attribute Details

#default_queue ⇒ Object (readonly)

Returns the value of attribute default_queue.



5
6
7
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 5

def default_queue
  @default_queue
end

#redis ⇒ Object (readonly)

Returns the value of attribute redis.



5
6
7
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 5

def redis
  @redis
end

Instance Method Details

#[](name) ⇒ Object 



14
15
16
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 14

def [](name)
  queues[name]
end

#active_connections ⇒ Object 



35
36
37
38
39
40
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 35

def active_connections
  redis.call!('CLIENT', 'LIST').split("\n").map do |connection_info|
    name_attr = connection_info.split(' ').detect { |a| a.match(/name=/) }
    name_attr ? name_attr[5..-1] : nil
  end.uniq.compact.reject(&:empty?)
end

#asynchronic? ⇒ Boolean

Returns:

  • (Boolean) 


31
32
33
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 31

def asynchronic?
  true
end

#clear ⇒ Object 



22
23
24
25
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 22

def clear
  queues.clear
  redis.call!('KEYS', 'ost:*').each { |k| redis.call!('DEL', k) }
end

#listener ⇒ Object 



27
28
29
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 27

def listener
  Listener.new
end

#queue_names ⇒ Object 



18
19
20
 
# File 'lib/asynchronic/queue_engine/ost.rb', line 18

def queue_names
  (queues.values.map(&:key) | redis.call!('KEYS', 'ost:*')).map { |q| q.to_s[4..-1].to_sym }
end