use plugin.stop to return from run

lib/logstash/inputs/heartbeat.rb

@@ -45,13 +45,14 @@ def register
 
   def run(queue)
     sequence = 0
+    @thread = Thread.current
 
     Stud.interval(@interval) do
       sequence += 1
       event = generate_message(sequence)
       decorate(event)
       queue << event
-      break if sequence == @count
+      break if sequence == @count || stop?
     end # loop
 
   end # def run
@@ -67,4 +68,7 @@ def generate_message(sequence)
     end
   end
 
+  def stop
+    Stud.stop!(@thread)
+  end
 end # class LogStash::Inputs::Heartbeat

spec/inputs/heartbeat_spec.rb

@@ -2,6 +2,11 @@
 require "logstash/inputs/heartbeat"
 
 describe LogStash::Inputs::Heartbeat do
+
+  it_behaves_like "an interruptible input plugin" do
+    let(:config) { { "interval" => 100 } }
+  end
+
   sequence = 1
   context "Default message test" do
     subject { LogStash::Inputs::Heartbeat.new({}) }
@@ -37,22 +42,14 @@
     end # it "should return an event with the current time (as epoch)"
   end # context "Epoch test"
 
-  it "should generate a fixed number of events then stop" do
-    count = 4
-    config = <<-CONFIG
-      input {
-        heartbeat {
-          interval => 1
-          message => "sequence"
-          count => #{count}
-        }
-      }
-    CONFIG
-
-    events = input(config) do |pipeline, queue|
-      count.times.map{queue.pop}
-    end
+  context "with a fixed count" do
+    let(:events) { [] }
+    let(:count) { 4 }
+    subject { LogStash::Inputs::Heartbeat.new("interval" => 1, "message" => "sequence", "count" => count) }
 
-    events.each_with_index{|event, i| expect(event['clock']).to eq(i + 1)}
+    it "should generate a fixed number of events then stop" do
+      subject.run(events)
+      events.each_with_index{|event, i| expect(event['clock']).to eq(i + 1)}
+    end
   end
 end