Class: Broadcaster

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

Defined Under Namespace

Classes: Watchdog

Constant Summary collapse

VERSION = 
'1.1.0'

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Broadcaster

Returns a new instance of Broadcaster.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/broadcaster.rb', line 22

def initialize(options={})
  @id = options.fetch(:id, SecureRandom.uuid)
  @logger = options.fetch(:logger, Broadcaster.logger)
  @logger_name = "Broadcaster (#{@id})"
  @redis_client = options.fetch(:redis_client, Broadcaster.redis_client)
  @redis_settings = options.fetch(:redis_settings, Broadcaster.redis_settings)
  @publisher = establish_connection
  @subscriptions = {}
  @mutex = Mutex.new
  @watchdog = Watchdog.new self, logger: logger,
                                 interval: options.fetch(:watchdog_interval, Broadcaster.watchdog_interval),
                                 timeout: options.fetch(:watchdog_timeout, Broadcaster.watchdog_timeout)

  start_listener
  watchdog.start
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



20
21
22
 
# File 'lib/broadcaster.rb', line 20

def id
  @id
end

#last_notification_at ⇒ Object (readonly)

Returns the value of attribute last_notification_at.



20
21
22
 
# File 'lib/broadcaster.rb', line 20

def last_notification_at
  @last_notification_at
end

Instance Method Details

#publish(channel, message) ⇒ Object 



39
40
41
42
 
# File 'lib/broadcaster.rb', line 39

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

#restart_listener ⇒ Object 



73
74
75
76
77
78
79
80
 
# File 'lib/broadcaster.rb', line 73

def restart_listener
  logger.warn(logger_name) { 'Restarting listener' }

  listener.kill if listener && listener != Thread.current
  disconnect subscriber

  start_listener
end

#shutdown ⇒ Object 



82
83
84
85
86
87
88
89
90
 
# File 'lib/broadcaster.rb', line 82

def shutdown
  watchdog.stop
  listener.kill

  disconnect subscriber
  disconnect publisher

  logger.info(logger_name) { 'Shutdown' }
end

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



44
45
46
47
48
49
50
51
52
 
# File 'lib/broadcaster.rb', line 44

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

#unsubscribe(subscription_id) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/broadcaster.rb', line 54

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

#unsubscribe_all ⇒ Object 



66
67
68
69
70
71
 
# File 'lib/broadcaster.rb', line 66

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