diff --git a/lib/logstash/filters/metrics.rb b/lib/logstash/filters/metrics.rb
index b771601..09ace94 100644
--- a/lib/logstash/filters/metrics.rb
+++ b/lib/logstash/filters/metrics.rb
@@ -150,6 +150,9 @@ class LogStash::Filters::Metrics < LogStash::Filters::Base
   # The percentiles that should be measured and emitted for timer values.
   config :percentiles, :validate => :array, :default => [1, 5, 10, 90, 95, 99, 100]
 
+  # If the metrics should be split into separate events
+  config :split_metrics, :validate => :boolean, :default => false
+
   def register
     require "metriks"
     require "socket"
@@ -183,6 +186,20 @@ def filter(event)
     end
   end # def filter
 
+  def make_event()
+    event = LogStash::Event.new
+    event.set("message", Socket.gethostname)
+    event
+  end # def make_event
+
+  def get_event(events)
+    if split_metrics or events.empty?
+      events << make_event
+    end
+
+    events.last
+  end # def get_event
+
   def flush(options = {})
     # Add 5 seconds to @last_flush and @last_clear counters
     # since this method is called every 5 seconds.
@@ -192,14 +209,14 @@ def flush(options = {})
     # Do nothing if there's nothing to do ;)
     return unless should_flush?
 
-    event = LogStash::Event.new
-    event.set("message", Socket.gethostname)
+    events = []
     @metric_meters.each_pair do |name, metric|
-      flush_rates event, name, metric
+      flush_rates get_event(events), name, metric
       metric.clear if should_clear?
     end
 
     @metric_timers.each_pair do |name, metric|
+      event = get_event(events)
       flush_rates event, name, metric
       # These 4 values are not sliding, so they probably are not useful.
       event.set("[#{name}][min]", metric.min)
@@ -224,8 +241,10 @@ def flush(options = {})
       @metric_timers.clear
     end
 
-    filter_matched(event)
-    return [event]
+    events.each do |event|
+      filter_matched(event)
+    end
+    return events
   end
 
   # this is a temporary fix to enable periodic flushes without using the plugin config:
diff --git a/spec/filters/metrics_spec.rb b/spec/filters/metrics_spec.rb
index b5616cc..bafdce3 100644
--- a/spec/filters/metrics_spec.rb
+++ b/spec/filters/metrics_spec.rb
@@ -60,6 +60,40 @@
           insist { events.first.get("http_404")["count"] } == 1
         end
       end
+
+      context "[split_metrics] on the first flush" do
+        subject {
+          config = {"meter" => ["http_%{response}"], "split_metrics" => true}
+          filter = LogStash::Filters::Metrics.new config
+          filter.register
+          filter.filter LogStash::Event.new({"response" => 200})
+          filter.filter LogStash::Event.new({"response" => 200})
+          filter.filter LogStash::Event.new({"response" => 404})
+          filter.flush
+        }
+
+        it "should flush counts in separate messages" do
+          insist { subject.length } == 2
+          reject { subject.first.get("http_200") }.nil?
+          insist { subject.first.get("http_200")["count"] } == 2
+          reject { subject.last.get("http_404") }.nil?
+          insist { subject.last.get("http_404")["count"] } == 1
+        end
+
+        def insist_metric_has_field(event, name, metrics)
+          reject { event.get(name) }.nil?
+          metrics.each do |metric|
+            insist { event.get(name) }.include? metric
+          end
+        end
+
+        it "should include rates and percentiles in separate messages" do
+          insist { subject.length } == 2
+          metrics = ["rate_1m", "rate_5m", "rate_15m"]
+          insist_metric_has_field(subject.first, "http_200", metrics)
+          insist_metric_has_field(subject.last, "http_404", metrics)
+        end
+      end
     end
 
     context "when custom rates and percentiles are selected" do