Fix PG::ConnectionBad issue

lib/queue_classic.rb

@@ -49,15 +49,7 @@ def self.has_connection?
   end
 
   def self.default_conn_adapter
-    t = Thread.current
-    return t[:qc_conn_adapter] if t[:qc_conn_adapter]
-    adapter = if rails_connection_sharing_enabled?
-      ConnAdapter.new(ActiveRecord::Base.connection.raw_connection)
-    else
-      ConnAdapter.new
-    end
-
-    t[:qc_conn_adapter] = adapter
+    Thread.current[:qc_conn_adapter] ||= ConnAdapter.new(rails_connection_sharing_enabled?)
   end
 
   def self.default_conn_adapter=(conn)

lib/queue_classic/conn_adapter.rb

@@ -4,24 +4,31 @@
 module QC
   class ConnAdapter
 
-    attr_accessor :connection
-    def initialize(c=nil)
-      @connection = c.nil? ? establish_new : validate!(c)
+    def initialize(active_record_connection_share)
+      @active_record_connection_share = active_record_connection_share
       @mutex = Mutex.new
     end
 
+    def connection
+      if @active_record_connection_share
+        ActiveRecord::Base.connection.raw_connection
+      else
+        @_connection ||= establish_new
+      end
+    end
+
     def execute(stmt, *params)
       @mutex.synchronize do
         QC.log(:at => "exec_sql", :sql => stmt.inspect)
         begin
           params = nil if params.empty?
-          r = @connection.exec(stmt, params)
+          r = connection.exec(stmt, params)
           result = []
           r.each {|t| result << t}
           result.length > 1 ? result : result.pop
         rescue PG::Error => e
           QC.log(:error => e.inspect)
-          @connection.reset
+          connection.reset
           raise
         end
       end
@@ -30,18 +37,18 @@ def execute(stmt, *params)
     def wait(time, *channels)
       @mutex.synchronize do
         listen_cmds = channels.map {|c| 'LISTEN "' + c.to_s + '"'}
-        @connection.exec(listen_cmds.join(';'))
+        connection.exec(listen_cmds.join(';'))
         wait_for_notify(time)
         unlisten_cmds = channels.map {|c| 'UNLISTEN "' + c.to_s + '"'}
-        @connection.exec(unlisten_cmds.join(';'))
+        connection.exec(unlisten_cmds.join(';'))
         drain_notify
       end
     end
 
     def disconnect
       @mutex.synchronize do
         begin
-          @connection.close
+          connection.close
         rescue => e
           QC.log(:at => 'disconnect', :error => e.message)
         end
@@ -59,12 +66,12 @@ def server_version
 
     def wait_for_notify(t)
       Array.new.tap do |msgs|
-        @connection.wait_for_notify(t) {|event, pid, msg| msgs << msg}
+        connection.wait_for_notify(t) {|event, pid, msg| msgs << msg}
       end
     end
 
     def drain_notify
-      until @connection.notifies.nil?
+      until connection.notifies.nil?
         QC.log(:at => "drain_notifications")
       end
     end