Add LISTEN support for JRuby / JDBC

bensheldon · Jan 6, 2022 · 8a6e5ca · 8a6e5ca
1 parent bf5d512
commit 8a6e5ca
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 13 deletions.
diff --git a/Gemfile.lock b/Gemfile.lock
@@ -158,27 +158,31 @@ GEM
     erubi (1.10.0)
     et-orbi (1.2.6)
       tzinfo
-    faraday (1.8.0)
+    faraday (1.9.2)
       faraday-em_http (~> 1.0)
       faraday-em_synchrony (~> 1.0)
       faraday-excon (~> 1.1)
-      faraday-httpclient (~> 1.0.1)
+      faraday-httpclient (~> 1.0)
+      faraday-multipart (~> 1.0)
       faraday-net_http (~> 1.0)
-      faraday-net_http_persistent (~> 1.1)
+      faraday-net_http_persistent (~> 1.0)
       faraday-patron (~> 1.0)
       faraday-rack (~> 1.0)
-      multipart-post (>= 1.2, < 3)
+      faraday-retry (~> 1.0)
       ruby2_keywords (>= 0.0.4)
     faraday-em_http (1.0.0)
     faraday-em_synchrony (1.0.0)
     faraday-excon (1.1.0)
     faraday-http-cache (2.2.0)
       faraday (>= 0.8)
     faraday-httpclient (1.0.1)
+    faraday-multipart (1.0.1)
+      multipart-post (>= 1.2, < 3)
     faraday-net_http (1.0.1)
     faraday-net_http_persistent (1.2.0)
     faraday-patron (1.0.0)
     faraday-rack (1.0.0)
+    faraday-retry (1.0.2)
     ffi (1.15.4)
     ffi (1.15.4-java)
     fiber-local (1.0.0)

diff --git a/lib/good_job/notifier.rb b/lib/good_job/notifier.rb
@@ -16,9 +16,6 @@ class Notifier
 
     include Notifier::ProcessRegistration
 
-    # Raised if the Database adapter does not implement LISTEN.
-    AdapterCannotListenError = Class.new(StandardError)
-
     # Default Postgres channel for LISTEN/NOTIFY
     CHANNEL = 'good_job'
     # Defaults for instance of Concurrent::ThreadPoolExecutor
@@ -129,8 +126,6 @@ def restart(timeout: -1)
     # @!visibility private
     # @return [void]
     def listen_observer(_time, _result, thread_error)
-      return if thread_error.is_a? AdapterCannotListenError
-
       if thread_error
         GoodJob._on_thread_error(thread_error)
         ActiveSupport::Notifications.instrument("notifier_notify_error.good_job", { error: thread_error })
@@ -214,6 +209,15 @@ def wait_for_notify
         raw_connection.wait_for_notify(WAIT_INTERVAL) do |channel, _pid, payload|
           yield(channel, payload)
         end
+      elsif raw_connection.respond_to?(:jdbc_connection)
+        raw_connection.execute_query("SELECT 1")
+        notifications = raw_connection.jdbc_connection.getNotifications
+        Array(notifications).each do |notification|
+          channel = notification.getName
+          payload = notification.getParameter
+          yield(channel, payload)
+        end
+        sleep WAIT_INTERVAL
       else
         sleep WAIT_INTERVAL
       end

diff --git a/spec/lib/good_job/notifier_spec.rb b/spec/lib/good_job/notifier_spec.rb
@@ -19,7 +19,7 @@
     end
   end
 
-  describe '#listen', skip_if_java: true do
+  describe '#listen' do
     it 'loops until it receives a command' do
       stub_const 'RECEIVED_MESSAGE', Concurrent::AtomicBoolean.new(false)
 
@@ -42,10 +42,11 @@
 
       notifier = described_class.new
       sleep_until(max: 5, increments_of: 0.5) { notifier.listening? }
+
       described_class.notify(true)
-      notifier.shutdown
+      wait_until(max: 5, increments_of: 0.5) { expect(on_thread_error).to have_received(:call).at_least(:once).with instance_of(ExpectedError) }
 
-      expect(on_thread_error).to have_received(:call).at_least(:once).with instance_of(ExpectedError)
+      notifier.shutdown
     end
   end
 

diff --git a/spec/test_app/config/initializers/good_job.rb b/spec/test_app/config/initializers/good_job.rb
@@ -14,7 +14,7 @@
 
   GoodJob.retry_on_unhandled_error = false
   GoodJob.preserve_job_records = true
-  GoodJob.on_thread_error = -> (error) { Rails.logger.warn(error) }
+  GoodJob.on_thread_error = -> (error) { Rails.logger.warn("#{error}\n#{error.backtrace}") }
 
   Rails.application.configure do
     config.good_job.enable_cron = ActiveModel::Type::Boolean.new.cast(ENV.fetch('GOOD_JOB_ENABLE_CRON', true))