Class: Marvin::Distributed::Handler

Inherits:
Base
  • Object
show all
Defined in:
lib/marvin/distributed/handler.rb

Constant Summary collapse

EVENT_WHITELIST = 
[:incoming_message, :incoming_action]
QUEUE_PROCESSING_SPACING = 
3

Instance Attribute Summary collapse

Attributes inherited from Base

#client, #from, #options, #target

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#_handle, #action, #addressed?, #ctcp, event_handlers_for, #from_channel?, #from_user?, #handle_incoming_numeric, #msg, on_event, on_numeric, #pm, #registered=, registered=, registered?, reloaded!, reloading!, #reply

Constructor Details

#initialize ⇒ Handler

Returns a new instance of Handler.



11
12
13
14
 
# File 'lib/marvin/distributed/handler.rb', line 11

def initialize
  super
  @message_queue = []
end

Instance Attribute Details

#message_queue ⇒ Object

Returns the value of attribute message_queue.



9
10
11
 
# File 'lib/marvin/distributed/handler.rb', line 9

def message_queue
  @message_queue
end

Class Method Details

.register!(*args) ⇒ Object 



74
75
76
77
78
79
 
# File 'lib/marvin/distributed/handler.rb', line 74

def register!(*args)
  # DO NOT register if this is  not a normal client.
  return unless Marvin::Loader.client?
  logger.info "Registering distributed handler on #{Marvin::Settings.client}"
  super
end

.whitelist_event(name) ⇒ Object 



69
70
71
72
 
# File 'lib/marvin/distributed/handler.rb', line 69

def whitelist_event(name)
  EVENT_WHITELIST << name.to_sym
  EVENT_WHITELIST.uniq!
end

Instance Method Details

#check_queue_progress ⇒ Object 



58
59
60
61
62
63
64
65
 
# File 'lib/marvin/distributed/handler.rb', line 58

def check_queue_progress
  if @message_queue.blank? && running?
    @running_timer.cancel
    @running_timer = nil
  elsif @message_queue.present? && !running?
    run!
  end
end

#dispatch(name, options, client = self.client) ⇒ Object 



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

def dispatch(name, options, client = self.client)
  return if client.blank?
  server = Marvin::Distributed::Server.next
  if server.blank?
    logger.debug "Distributed handler is currently busy - adding to queue"
    # TODO: Add to queued messages, wait
    @message_queue << [name, options, client]
    run! unless running?
  else
    server.dispatch(client, name, options)
  end
rescue Exception => e
  logger.warn "Error dispatching #{name}"
  Marvin::ExceptionTracker.log(e)
end

#handle(message, options) ⇒ Object 



16
17
18
19
20
 
# File 'lib/marvin/distributed/handler.rb', line 16

def handle(message, options)
  return unless EVENT_WHITELIST.include?(message)
  super(message, options)
  dispatch(message, options)
end

#process_queue ⇒ Object 



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

def process_queue
  count = [@message_queue.size, Server.free_connections.size].min
  logger.debug "Processing #{count} item(s) from the message queue"
  count.times { |item| dispatch(*@message_queue.shift) }
  if @message_queue.empty?
    logger.debug "The message queue is now empty"
  else
    logger.debug "The message queue still has #{count} item(s)"
  end
  check_queue_progress
end

#run! ⇒ Object 



54
55
56
 
# File 'lib/marvin/distributed/handler.rb', line 54

def run!
  @running_timer = EventMachine::PeriodicTimer.new(QUEUE_PROCESSING_SPACING) { process_queue }
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


50
51
52
 
# File 'lib/marvin/distributed/handler.rb', line 50

def running?
  @running_timer.present?
end