Class: PgVersions::ConnectionThread

Inherits:

Object

• Object
• PgVersions::ConnectionThread

Defined in:

lib/pg_versions/pg_versions.rb

Defined Under Namespace

Classes: Closing

Instance Attribute Summary

• #status ⇒ Object readonly

  Returns the value of attribute status.

Instance Method Summary

• #initialize(connection) ⇒ ConnectionThread constructor

  A new instance of ConnectionThread.

• #request(function, *params) ⇒ Object
• #request_nonblock(function, *params) ⇒ Object

Constructor Details

#initialize(connection) ⇒ ConnectionThread

Returns a new instance of ConnectionThread.

# File 'lib/pg_versions/pg_versions.rb', line 170

def initialize(connection)
  retry_on_exceptions = [ PG::ConnectionBad, PG::UnableToSend ]
  retry_on_exceptions << ActiveRecord::ConnectionNotEstablished  if defined? ActiveRecord

  @subscribers = Hash.new { |h,k| h[k] = Set.new }
  @status = :disconnected

  @thread_requests_mutex = Mutex.new
  @thread_requests = []
  thread_requests_notify_r, @thread_requests_notify_w = IO.pipe

  @thread = Thread.new {
    reset_connection = false
    retry_delay = 0
    begin
      PgVersions.with_connection(connection, reset: reset_connection) do |pg_connection|
        
        @status = :connected
        retry_delay = 0

        @subscribers.each_key { |channel|
          listen(pg_connection, channel)
        }
        read(pg_connection, @subscribers.keys).each_pair { |channel, version|
          @subscribers[channel].each { |subscriber|
            subscriber.notify({ channel => version })
          }
        }

        loop {
          @thread_requests_mutex.synchronize { 
            @thread_requests.each { |function, retpipe, params|
              raise Closing, retpipe  if function == :stop
              retpipe << send(function, pg_connection, *params)
            }
            @thread_requests.clear
          }

          while notification = pg_connection.notifies
            channel, payload = notification[:relname], notification[:extra]
            @subscribers[channel].each { |subscriber|
              subscriber.notify({ channel => PgVersions.string_to_version(payload) })
            }
          end

          #TODO: handle errors
          reads,_writes,_errors = IO::select([pg_connection.socket_io, thread_requests_notify_r])
          
          if reads.include?(pg_connection.socket_io)
            pg_connection.consume_input
          end

          if reads.include?(thread_requests_notify_r)
            thread_requests_notify_r.read(1)
          end
        }
      rescue Closing => e
        e.retpipe << true
      end
    rescue *retry_on_exceptions => e
      reset_connection = true
      @status = :disconnected
      $stderr.puts "Pg connection failed. Retrying in #{retry_delay/1000.0}s."
      sleep retry_delay/1000.0
      retry_delay = { 0=>100, 100=>1000, 1000=>2000, 2000=>2000 }[retry_delay]
      retry
    end
  } 
  @thread.abort_on_exception = true
end

Instance Attribute Details

#status ⇒ Object (readonly)

Returns the value of attribute status.

# File 'lib/pg_versions/pg_versions.rb', line 168

def status
  @status
end

Instance Method Details

#request(function, *params) ⇒ Object

# File 'lib/pg_versions/pg_versions.rb', line 277

def request(function, *params)
  request_nonblock(function, *params).pop
end

#request_nonblock(function, *params) ⇒ Object

# File 'lib/pg_versions/pg_versions.rb', line 281

def request_nonblock(function, *params)
  retpipe = Queue.new
  @thread_requests_mutex.synchronize {
    @thread_requests.push [function, retpipe, params]
  }
  @thread_requests_notify_w.write('!')
  retpipe
end