gelf.rb in logstash-input-gelf-2.0.0

- old
+ new

@@ -1,11 +1,12 @@
 # encoding: utf-8
-require "date"
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "logstash/json"
 require "logstash/timestamp"
+require "stud/interval"
+require "date"
 require "socket"
 
 # This input will read GELF messages as events over the network,
 # making it a good choice if you already use Graylog2 today.
 #
@@ -40,16 +41,16 @@
   #
   # e.g. `\_foo` becomes `foo`
   #
   config :strip_leading_underscore, :validate => :boolean, :default => true
 
+  RECONNECT_BACKOFF_SLEEP = 5
+
   public
   def initialize(params)
     super
     BasicSocket.do_not_reverse_lookup = true
-    @shutdown_requested = false
-    @udp = nil
   end # def initialize
 
   public
   def register
     require 'gelfd'
@@ -58,46 +59,35 @@
   public
   def run(output_queue)
     begin
       # udp server
       udp_listener(output_queue)
-    rescue LogStash::ShutdownSignal
-      @shutdown_requested = true
     rescue => e
-      unless @shutdown_requested
+      unless stop?
         @logger.warn("gelf listener died", :exception => e, :backtrace => e.backtrace)
-        sleep(5)
-        retry
+        Stud.stoppable_sleep(RECONNECT_BACKOFF_SLEEP) { stop? }
+        retry unless stop?
       end
     end # begin
   end # def run
 
   public
-  def teardown
-    @shutdown_requested = true
-    if @udp
-      @udp.close_read rescue nil
-      @udp.close_write rescue nil
-      @udp = nil
-    end
-    finished
+  def stop
+    @udp.close
+  rescue IOError # the plugin is currently shutting down, so its safe to ignore theses errors
   end
 
   private
   def udp_listener(output_queue)
     @logger.info("Starting gelf listener", :address => "#{@host}:#{@port}")
 
-    if @udp
-      @udp.close_read rescue nil
-      @udp.close_write rescue nil
-    end
-
     @udp = UDPSocket.new(Socket::AF_INET)
     @udp.bind(@host, @port)
 
-    while !@shutdown_requested
+    while !stop?
       line, client = @udp.recvfrom(8192)
+
       begin
         data = Gelfd::Parser.parse(line)
       rescue => ex
         @logger.warn("Gelfd failed to parse a message skipping", :exception => ex, :backtrace => ex.backtrace)
         next
@@ -111,9 +101,10 @@
       event["source_host"] = client[3]
       if event["timestamp"].is_a?(Numeric)
         event.timestamp = LogStash::Timestamp.at(event["timestamp"])
         event.remove("timestamp")
       end
+
       remap_gelf(event) if @remap
       strip_leading_underscore(event) if @strip_leading_underscore
       decorate(event)
 
       output_queue << event