Make usage of the new stop semantics to properly shutdown the plugin

lib/logstash/inputs/zeromq.rb

@@ -77,8 +77,6 @@ def register
     require "logstash/util/zeromq"
     self.class.send(:include, LogStash::Util::ZeroMQ)
 
-    @run_mutex = Mutex.new
-    @run       = true
     case @topology
     when "pair"
       zmq_const = ZMQ::PAIR 
@@ -116,7 +114,6 @@ def register
   end # def register
 
   def close
-    @run_mutex.synchronize{@run = false}
     error_check(@zsocket.close, "while closing the zmq socket")
     context.terminate
   end # def close
@@ -134,6 +131,8 @@ def run(output_queue)
         m1 = ""
         rc = @zsocket.recv_string(m1, ZMQ::DONTWAIT)
         next if rc == -1 && ZMQ::Util.errno == ZMQ::EAGAIN
+        error_check(rc, "in recv_string")
+
         @logger.debug("ZMQ receiving", :event => m1)
         msg = m1
         # If we have more parts, we'll eat the first as the topic
@@ -152,9 +151,6 @@ def run(output_queue)
           output_queue << event
         end
       end
-    rescue LogStash::ShutdownSignal
-      # shutdown
-      return
     rescue => e
       @logger.debug("ZMQ Error", :subscriber => @zsocket,
                     :exception => e)
@@ -167,8 +163,4 @@ def build_source_string
     id = @address.first.clone
   end
 
-  def stop?
-    !@run
-  end
-
 end # class LogStash::Inputs::ZeroMQ

lib/logstash/util/zeromq.rb

@@ -3,13 +3,12 @@
 require "logstash/namespace"
 
 module LogStash::Util::ZeroMQ
-  CONTEXT = ZMQ::Context.new
   # LOGSTASH-400
   # see https://github.com/chuckremes/ffi-rzmq/blob/master/lib/ffi-rzmq/socket.rb#L93-117
   STRING_OPTS = %w{IDENTITY SUBSCRIBE UNSUBSCRIBE}
 
   def context
-    CONTEXT
+    @context ||= ZMQ::Context.new
   end
 
   def setup(socket, address)

spec/inputs/zeromq_spec.rb

@@ -14,6 +14,14 @@
       expect { plugin.close }.to_not raise_error
     end
 
-  end
+    context "when interrupting the plugin" do
+      it_behaves_like "an interruptible input plugin" do
+        let(:config) { { "topology" => "pushpull" } }
+        after do
+          subject.close
+        end
+      end
+    end
 
+  end
 end

spec/spec_helper.rb

@@ -23,7 +23,7 @@ def input(config, size, &block)
     result = size.times.inject([]) do |acc|
       acc << queue.pop
     end
-    plugin.close
+    plugin.do_stop
     pipeline_thread.join
     result
   end # def input