Class: Broadcaster

Inherits:
Object
  • Object
show all
Extended by:
ClassConfig
Defined in:
lib/broadcaster.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Broadcaster

Returns a new instance of Broadcaster.



14
15
16
17
18
19
20
21
22
 
# File 'lib/broadcaster.rb', line 14

def initialize(options={})
  @id = options.fetch(:id, SecureRandom.uuid)
  @logger = options.fetch(:logger, Broadcaster.logger)
  @publisher = Redic.new options.fetch(:redis_url, Broadcaster.redis_url)
  @subscriber = Redic.new options.fetch(:redis_url, Broadcaster.redis_url)
  @subscriptions = Hash.new { |h,k| h[k] = {} }
  @mutex = Mutex.new
  listen
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



12
13
14
 
# File 'lib/broadcaster.rb', line 12

def id
  @id
end

Instance Method Details

#publish(channel, message) ⇒ Object 



24
25
26
27
 
# File 'lib/broadcaster.rb', line 24

def publish(channel, message)
  logger.debug(self.class) { "Published | #{scoped(channel)} | #{message}" }
  publisher.call 'PUBLISH', scoped(channel), Marshal.dump(message)
end

#subscribe(channel, callable = nil, &block) ⇒ Object 



29
30
31
32
33
34
35
36
 
# File 'lib/broadcaster.rb', line 29

def subscribe(channel, callable=nil, &block)
  mutex.synchronize do
    SecureRandom.uuid.tap do |subscription_id|
      logger.debug(self.class) { "Subscribed | #{scoped(channel)} | #{subscription_id}" }
      subscriptions[scoped(channel)][subscription_id] = callable || block
    end
  end
end

#unsubscribe(subscription_id) ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/broadcaster.rb', line 38

def unsubscribe(subscription_id)
  mutex.synchronize do
    channel, _ = subscriptions.detect { |k,v| v.key? subscription_id }
    if channel
      logger.debug(self.class) { "Unsubscribed | #{channel} | #{subscription_id}" }
      block = subscriptions[channel].delete subscription_id
      subscriptions.delete_if { |k,v| v.empty? }
      block
    end
  end
end

#unsubscribe_all ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/broadcaster.rb', line 50

def unsubscribe_all
  mutex.synchronize do
    logger.debug(self.class) { 'Unsubscribed all' }
    subscriptions.clear
  end
end