aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--lib/mauve/configuration_builders/server.rb7
-rw-r--r--lib/mauve/notifier.rb19
-rw-r--r--lib/mauve/processor.rb82
-rw-r--r--lib/mauve/server.rb50
4 files changed, 103 insertions, 55 deletions
diff --git a/lib/mauve/configuration_builders/server.rb b/lib/mauve/configuration_builders/server.rb
index e3654b9..feb75b8 100644
--- a/lib/mauve/configuration_builders/server.rb
+++ b/lib/mauve/configuration_builders/server.rb
@@ -163,6 +163,13 @@ module Mauve
# The period of sleep during which no heartbeats are raised.
#
is_attribute "initial_sleep"
+
+ #
+ # The next two attributes determine if packet/notitication bufferes are
+ # used. These both default to "true"
+ #
+ is_attribute "use_packet_buffer"
+ is_attribute "use_notification_buffer"
def builder_setup
@result = Mauve::Server.instance
diff --git a/lib/mauve/notifier.rb b/lib/mauve/notifier.rb
index 8a26b2c..c473acf 100644
--- a/lib/mauve/notifier.rb
+++ b/lib/mauve/notifier.rb
@@ -31,6 +31,18 @@ module Mauve
end
end
+
+ #
+ # This sends the notification for an alert
+ #
+ def notify(alert, at)
+ if alert.alert_group.nil?
+ logger.warn "Could not notify for #{alert} since there are no matching alert groups"
+ else
+ alert.alert_group.notify(alert, at)
+ end
+ end
+
private
@@ -86,12 +98,7 @@ module Mauve
# Empty the buffer, one notification at a time.
#
sz.times do
- alert, at = Server.notification_pop
- if alert.alert_group.nil?
- logger.warn "Could not notify for #{alert} since there are no matching alert groups"
- else
- alert.alert_group.notify(alert, at)
- end
+ notify(*Server.notification_pop)
end
end
diff --git a/lib/mauve/processor.rb b/lib/mauve/processor.rb
index 9896530..0ac7b59 100644
--- a/lib/mauve/processor.rb
+++ b/lib/mauve/processor.rb
@@ -78,6 +78,44 @@ module Mauve
#
do_processor
end
+
+ # This processes an incoming packet. It is in a seperate method so it can
+ # be (de)coupled as needed from the UDP server.
+ #
+ def process_packet(data, client, received_at)
+ #
+ # Uh-oh. Nil data? That's craaaazy
+ #
+ return nil if data.nil?
+
+ ip_source = "#{client[3]}"
+ update = Proto::AlertUpdate.new
+
+ update.parse_from_string(data)
+
+ if @transmission_id_cache[update.transmission_id.to_s]
+ logger.debug("Ignoring duplicate transmission id #{update.transmission_id}")
+ return nil
+ end
+
+ logger.debug "Update #{update.transmission_id} sent at #{update.transmission_time} received at #{received_at.to_i} from "+
+ "'#{update.source}'@#{ip_source} alerts #{update.alert.length}"
+
+ Alert.receive_update(update, received_at, ip_source)
+
+ rescue Protobuf::InvalidWireType,
+ NotImplementedError,
+ DataObjects::IntegrityError => ex
+
+ logger.error "#{ex} (#{ex.class}) while parsing #{data.length} bytes "+
+ "starting '#{data[0..15].inspect}' from #{ip_source}"
+
+ logger.debug ex.backtrace.join("\n")
+
+ ensure
+ @transmission_id_cache[update.transmission_id.to_s] = Time.now
+ end
+
private
@@ -157,52 +195,12 @@ module Mauve
sz = Server.packet_buffer_size
sz.times do
- data, client, received_at = Server.packet_pop
-
- #
- # Uh-oh. Nil data? That's craaaazy
- #
- next if data.nil?
-
-
- # logger.debug("Got #{data.inspect} from #{client.inspect}")
-
- ip_source = "#{client[3]}"
- update = Proto::AlertUpdate.new
-
- begin
- update.parse_from_string(data)
-
- if @transmission_id_cache[update.transmission_id.to_s]
- logger.debug("Ignoring duplicate transmission id #{update.transmission_id}")
- #
- # Continue with next packet.
- #
- next
- end
-
- logger.debug "Update #{update.transmission_id} sent at #{update.transmission_time} received at #{received_at.to_i} from "+
- "'#{update.source}'@#{ip_source} alerts #{update.alert.length}"
-
- Alert.receive_update(update, received_at, ip_source)
-
- rescue Protobuf::InvalidWireType,
- NotImplementedError,
- DataObjects::IntegrityError => ex
-
- logger.error "#{ex} (#{ex.class}) while parsing #{data.length} bytes "+
- "starting '#{data[0..15].inspect}' from #{ip_source}"
-
- logger.debug ex.backtrace.join("\n")
-
- ensure
- @transmission_id_cache[update.transmission_id.to_s] = Time.now
- end
+ process_packet(*Server.packet_pop)
end
end
def timer_should_stop?
- (Server.packet_buffer_size > 0)
+ (Server.packet_buffer_size > 0) or self.should_stop?
end
end
diff --git a/lib/mauve/server.rb b/lib/mauve/server.rb
index 3e82858..db5fda6 100644
--- a/lib/mauve/server.rb
+++ b/lib/mauve/server.rb
@@ -40,7 +40,7 @@ module Mauve
@started_at = Time.now
@initial_sleep = 300
-
+
#
# Keep these queues here to prevent a crash in a subthread losing all the
# subsquent things in the queue.
@@ -69,7 +69,38 @@ module Mauve
raise ArgumentError, "database must be a string" unless d.is_a?(String)
@database = d
end
-
+
+ #
+ # Sets up the packet buffer (or not). The argument can be "false" or "no"
+ # or a FalseClass object for no. Anything else makes no change.
+ #
+ # @param [String] arg
+ # @return [Array or nil]
+ def use_packet_buffer=(arg)
+ logger.debug(arg)
+ if arg.is_a?(FalseClass) or arg =~ /^(n(o)?|f(alse)?)$/i
+ @packet_buffer = nil
+ end
+
+ @packet_buffer
+ end
+
+ #
+ # Sets up the notification buffer (or not). The argument can be "false" or
+ # "no" or a FalseClass object for no. Anything else makes no change.
+ #
+ # @param [String] arg
+ # @return [Array or nil]
+ def use_notification_buffer=(arg)
+ logger.debug(arg)
+ if arg.is_a?(FalseClass) or arg =~ /^(n(o)?|f(alse)?)$/i
+ @notification_buffer = nil
+ end
+
+ @notification_buffer
+ end
+
+
# Set the sleep period during which notifications about old alerts are
# suppressed.
#
@@ -97,11 +128,8 @@ module Mauve
# @return [NilClass]
def setup
#
+ # Set up the database
#
- #
- @packet_buffer = []
- @notification_buffer = []
-
DataMapper.setup(:default, @database)
# DataMapper.logger = Log4r::Logger.new("Mauve::DataMapper")
@@ -261,6 +289,8 @@ module Mauve
# @param [String] a Packet from the UDP server
def packet_enq(a)
instance.packet_buffer.push(a)
+ rescue NoMethodError
+ Processor.instance.process_packet(*a)
end
# Shift a packet off the front of the +packet buffer+
@@ -275,6 +305,8 @@ module Mauve
# @return [Integer}
def packet_buffer_size
instance.packet_buffer.size
+ rescue NoMethodError
+ 0
end
alias packet_push packet_enq
@@ -285,6 +317,8 @@ module Mauve
# @param [Array] a Notification array, consisting of a Person and the args to Mauve::Person#send_alert
def notification_enq(a)
instance.notification_buffer.push(a)
+ rescue NoMethodError
+ Notifier.instance.notify(*a)
end
# Shift a notification off the front of the +notification_buffer+
@@ -299,8 +333,10 @@ module Mauve
# @return [Integer]
def notification_buffer_size
instance.notification_buffer.size
+ rescue NoMethodError
+ 0
end
-
+
alias notification_push notification_enq
alias notification_pop notification_deq