add code and test to set eviction_block

update changelog and version

Closes #16

Author: guyboertje

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.0.4
+ - Add constructional method to allow an eviction specific block to be set.
+
 ## 2.0.3
  - Add pseudo codec IdentityMapCodec.  Support class for identity based multiline processing.
 

lib/logstash/codecs/identity_map_codec.rb

@@ -101,7 +101,8 @@ def initialize(codec)
     @max_identities = MAX_IDENTITIES
     @evict_timeout = EVICT_TIMEOUT
     @cleaner = MapCleaner.new(self, CLEANER_INTERVAL)
-    @decode_block = lambda {|*| }
+    @decode_block = lambda {|*| true }
+    @eviction_block = nil
   end
 
   # ==============================================
@@ -126,6 +127,12 @@ def cleaner_interval(interval)
     @cleaner = MapCleaner.new(self, interval.to_i)
     self
   end
+
+  # used to add  a non-default eviction block
+  def eviction_block(block)
+    @eviction_block = block
+    self
+  end
   # end Constructional/builder methods
   # ==============================================
 
@@ -184,7 +191,7 @@ def map_cleanup
     # contents should not mutate during this call
     identity_map.delete_if do |identity, compo|
       if (flag = compo.timeout <= cut_off)
-        compo.codec.flush(&@decode_block)
+        compo.codec.flush(&(@eviction_block || @decode_block))
       end
       flag
     end

logstash-codec-multiline.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-codec-multiline'
-  s.version         = '2.0.3'
+  s.version         = '2.0.4'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "The multiline codec will collapse multiline messages and merge them into a single event."
   s.description     = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"
@@ -20,7 +20,7 @@ Gem::Specification.new do |s|
   s.metadata = { "logstash_plugin" => "true", "logstash_group" => "codec" }
 
   # Gem dependencies
-  s.add_runtime_dependency "logstash-core", ">= 2.0.0.beta2", "< 3.0.0"
+  s.add_runtime_dependency "logstash-core", ">= 2.0.0", "< 3.0.0"
 
   s.add_runtime_dependency 'logstash-patterns-core'
   s.add_runtime_dependency 'jls-grok', '~> 0.11.1'

spec/codecs/identity_map_codec_spec.rb

@@ -18,7 +18,7 @@ def initialize() @tracer = []; end
   def clone() self.class.new; end
   def decode(data) @tracer.push [:decode, data]; end
   def encode(event) @tracer.push [:encode, event]; end
-  def flush(&block) @tracer.push [:flush, true]; end
+  def flush(&block) @tracer.push [:flush, block.call]; end
   def close() @tracer.push [:close, true]; end
   def logger() @logger ||= LogTracer.new; end
 
@@ -175,7 +175,7 @@ def trace_for(symbol)
           sleep(1.2)
           demuxer.decode(Object.new, "stream2")
           expect(demuxer.identity_count).to eq(limit)
-          expect { demuxer.decode(Object.new, "stream4") }.not_to raise_error
+          expect { demuxer.decode(Object.new, "stream4"){|*| 42 } }.not_to raise_error
         end
       end
     end
@@ -195,12 +195,27 @@ def trace_for(symbol)
   end
 
   describe "codec eviction" do
-    let(:demuxer) { described_class.new(codec).evict_timeout(1).cleaner_interval(1) }
     context "when an identity has become stale" do
+      let(:demuxer) { described_class.new(codec).evict_timeout(1).cleaner_interval(1) }
       it "the cleaner evicts the codec and flushes it first" do
-        demuxer.decode(Object.new, "stream1")
+        demuxer.decode(Object.new, "stream1"){|*| 42}
+        sleep(2.1)
+        expect(codec.trace_for(:flush)).to eq(42)
+        expect(demuxer.identity_map.keys).not_to include("stream1")
+      end
+    end
+
+    context "when an identity has become stale and an evition block is set" do
+      let(:demuxer) do
+        described_class.new(codec)
+        .evict_timeout(1)
+        .cleaner_interval(1)
+        .eviction_block(lambda {|*| 24} )
+      end
+      it "the cleaner evicts the codec and flushes it first using the eviction_block" do
+        demuxer.decode(Object.new, "stream1"){|*| 42}
         sleep(2.1)
-        expect(codec.trace_for(:flush)).to be_truthy
+        expect(codec.trace_for(:flush)).to eq(24)
         expect(demuxer.identity_map.keys).not_to include("stream1")
       end
     end