Refactor: avoid usage of CHM (JRuby 9.3.4 work-around) (#248)

Author: kares

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.3.7
+  - Refactor: avoid usage of CHM (JRuby 9.3.4 work-around) [#248](https://github.com/logstash-plugins/logstash-output-s3/pull/248)
+
 ## 4.3.6
   - Docs: more documentation on restore + temp dir [#236](https://github.com/logstash-plugins/logstash-output-s3/pull/236)
     * minor logging improvements - use the same path: naming convention

lib/logstash/outputs/s3/file_repository.rb

@@ -1,6 +1,6 @@
 # encoding: utf-8
 require "java"
-require "concurrent"
+require "concurrent/map"
 require "concurrent/timer_task"
 require "logstash/util"
 
@@ -39,25 +39,26 @@ def delete!
         end
 
         class FactoryInitializer
-          include java.util.function.Function
+
           def initialize(tags, encoding, temporary_directory, stale_time)
             @tags = tags
             @encoding = encoding
             @temporary_directory = temporary_directory
             @stale_time = stale_time
           end
 
-          def apply(prefix_key)
+          def create_value(prefix_key)
             PrefixedValue.new(TemporaryFileFactory.new(prefix_key, @tags, @encoding, @temporary_directory), @stale_time)
           end
+
         end
 
         def initialize(tags, encoding, temporary_directory,
                        stale_time = DEFAULT_STALE_TIME_SECS,
                        sweeper_interval = DEFAULT_STATE_SWEEPER_INTERVAL_SECS)
           # The path need to contains the prefix so when we start
           # logtash after a crash we keep the remote structure
-          @prefixed_factories = java.util.concurrent.ConcurrentHashMap.new
+          @prefixed_factories = Concurrent::Map.new
 
           @sweeper_interval = sweeper_interval
 
@@ -67,18 +68,19 @@ def initialize(tags, encoding, temporary_directory,
         end
 
         def keys
-          @prefixed_factories.keySet
+          @prefixed_factories.keys
         end
 
         def each_files
-          @prefixed_factories.elements.each do |prefixed_file|
+          @prefixed_factories.values.each do |prefixed_file|
             prefixed_file.with_lock { |factory| yield factory.current }
           end
         end
 
         # Return the file factory
         def get_factory(prefix_key)
-          @prefixed_factories.computeIfAbsent(prefix_key, @factory_initializer).with_lock { |factory| yield factory }
+          prefix_val = @prefixed_factories.fetch_or_store(prefix_key) { @factory_initializer.create_value(prefix_key) }
+          prefix_val.with_lock { |factory| yield factory }
         end
 
         def get_file(prefix_key)
@@ -95,7 +97,7 @@ def size
 
         def remove_stale(k, v)
           if v.stale?
-            @prefixed_factories.remove(k, v)
+            @prefixed_factories.delete_pair(k, v)
             v.delete!
           end
         end
@@ -104,7 +106,7 @@ def start_stale_sweeper
           @stale_sweeper = Concurrent::TimerTask.new(:execution_interval => @sweeper_interval) do
             LogStash::Util.set_thread_name("S3, Stale factory sweeper")
 
-            @prefixed_factories.forEach{|k,v| remove_stale(k,v)}
+            @prefixed_factories.each { |k, v| remove_stale(k,v) }
           end
 
           @stale_sweeper.execute

logstash-output-s3.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-s3'
-  s.version         = '4.3.6'
+  s.version         = '4.3.7'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Sends Logstash events to the Amazon Simple Storage Service"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/outputs/s3/file_repository_spec.rb

@@ -89,7 +89,7 @@
   it "returns all available keys" do
     subject.get_file(prefix_key) { |file| file.write("something") }
     expect(subject.keys).to include(prefix_key)
-    expect(subject.keys.to_a.size).to eq(1)
+    expect(subject.keys.size).to eq(1)
   end
 
   it "clean stale factories" do
@@ -105,9 +105,14 @@
 
     @file_repository.get_file("another-prefix") { |file| file.write("hello") }
     expect(@file_repository.size).to eq(2)
+    sleep 1.2 # allow sweeper to kick in
     try(10) { expect(@file_repository.size).to eq(1) }
     expect(File.directory?(path)).to be_falsey
+
+    sleep 1.5 # allow sweeper to kick in, again
+    expect(@file_repository.size).to eq(1)
   end
+
 end