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 165

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|
					#p 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 163

def status
  @status
end

Instance Method Details

#request(function, *params) ⇒ Object

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

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

#request_nonblock(function, *params) ⇒ Object

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

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