class PG::Connection

def self.conndefaults_hash
	return self.conndefaults.each_with_object({}) do |info, hash|
		hash[ info[:keyword].to_sym ] = info[:val]
	end
end

def self.connect_hash_to_string( hash )
	hash.map { |k,v| "#{k}=#{quote_connstr(v)}" }.join( ' ' )
end

def self.parse_connect_args( *args )
	hash_arg = args.last.is_a?( Hash ) ? args.pop.transform_keys(&:to_sym) : {}
	iopts = {}
	if args.length == 1
		case args.first
		when URI, /=/, /:\/\//
			# Option or URL string style
			conn_string = args.first.to_s
			iopts = PG::Connection.conninfo_parse(conn_string).each_with_object({}){|h, o| o[h[:keyword].to_sym] = h[:val] if h[:val] }
		else
			# Positional parameters (only host given)
			iopts[CONNECT_ARGUMENT_ORDER.first.to_sym] = args.first
		end
	else
		# Positional parameters with host and more
		max = CONNECT_ARGUMENT_ORDER.length
		raise ArgumentError,
				"Extra positional parameter %d: %p" % [ max + 1, args[max] ] if args.length > max
		CONNECT_ARGUMENT_ORDER.zip( args ) do |(k,v)|
			iopts[ k.to_sym ] = v if v
		end
		iopts.delete(:tty) # ignore obsolete tty parameter
	end
	iopts.merge!( hash_arg )
	if !iopts[:fallback_application_name]
		iopts[:fallback_application_name] = $0.sub( /^(.{30}).{4,}(.{30})$/ ){ $1+"..."+$2 }
	end
	return connect_hash_to_string(iopts)
end

def self.quote_connstr( value )
	return "'" + value.to_s.gsub( /[\\']/ ) {|m| '\\' + m } + "'"
end

def async_api=(enable)
	self.async_send_api = enable
	REDIRECT_METHODS.each do |ali, (async, sync)|
		remove_method(ali) if method_defined?(ali)
		alias_method( ali, enable ? async : sync )
	end
	REDIRECT_CLASS_METHODS.each do |ali, (async, sync)|
		singleton_class.remove_method(ali) if method_defined?(ali)
		singleton_class.alias_method(ali, enable ? async : sync )
	end
end

def async_connect_or_reset(poll_meth)
 the progress of the connection, waiting for the socket to become readable/writable before polling it
eo = conninfo_hash[:connect_timeout].to_i) && timeo > 0
st timeout is 2 seconds - like in libpq
= [timeo, 2].max
ime = timeo + Process.clock_gettime(Process::CLOCK_MONOTONIC)
atus = PG::PGRES_POLLING_WRITING
oll_status == PG::PGRES_POLLING_OK ||
status == PG::PGRES_POLLING_FAILED
t = stop_time&.-(Process.clock_gettime(Process::CLOCK_MONOTONIC))
= if !timeout || timeout >= 0
the socket needs to read, wait 'til it becomes readable to poll again
poll_status
PG::PGRES_POLLING_READING
efined?(IO::READABLE) # ruby-3.0+
ket_io.wait(IO::READABLE | IO::PRIORITY, timeout)

select([socket_io], nil, [socket_io], timeout)
and the same for when the socket needs to write
PG::PGRES_POLLING_WRITING
efined?(IO::WRITABLE) # ruby-3.0+
se wait instead of wait_readable, since connection errors are delivered as
xceptional/priority events on Windows.
ket_io.wait(IO::WRITABLE | IO::PRIORITY, timeout)

o#wait on ruby-2.x doesn't wait for priority, so fallback to IO.select
select(nil, [socket_io], [socket_io], timeout)
ection to server at "localhost" (127.0.0.1), port 5433 failed: timeout expired (PG::ConnectionBad)
ection to server on socket "/var/run/postgresql/.s.PGSQL.5433" failed: No such file or directory
 event
lf.class.send(:host_is_named_pipe?, host)
host = "on socket \"#{host}\""
 respond_to?(:hostaddr)
host = "at \"#{host}\" (#{hostaddr}), port #{port}"
host = "at \"#{host}\", port #{port}"
 PG::ConnectionBad.new("connection to server #{connhost} failed: timeout expired", connection: self)
k to see if it's finished or failed yet
tatus = send( poll_meth )
status = status unless [PG::CONNECTION_BAD, PG::CONNECTION_OK].include?(status)
status == PG::CONNECTION_OK
error_message

PG::ConnectionBad.new(msg, connection: self)
onnection to nonblocking to handle all blocking states in ruby.
way a fiber scheduler is able to handle IO requests.
tnonblocking(true)
ush_data = true
ault_encoding

def async_send_api=(enable)
	REDIRECT_SEND_METHODS.each do |ali, (async, sync)|
		undef_method(ali) if method_defined?(ali)
		alias_method( ali, enable ? async : sync )
	end
end

def cancel
	be_pid = backend_pid
	be_key = backend_key
	cancel_request = [0x10, 1234, 5678, be_pid, be_key].pack("NnnNN")
	if Fiber.respond_to?(:scheduler) && Fiber.scheduler && RUBY_PLATFORM =~ /mingw|mswin/
		# Ruby's nonblocking IO is not really supported on Windows.
		# We work around by using threads and explicit calls to wait_readable/wait_writable.
		cl = Thread.new(socket_io.remote_address) { |ra| ra.connect }.value
		begin
			cl.write_nonblock(cancel_request)
		rescue IO::WaitReadable, Errno::EINTR
			cl.wait_writable
			retry
		end
		begin
			cl.read_nonblock(1)
		rescue IO::WaitReadable, Errno::EINTR
			cl.wait_readable
			retry
		rescue EOFError
		end
	elsif RUBY_ENGINE == 'truffleruby'
		begin
			cl = socket_io.remote_address.connect
		rescue NotImplementedError
			# Workaround for truffleruby < 21.3.0
			cl2 = Socket.for_fd(socket_io.fileno)
			cl2.autoclose = false
			adr = cl2.remote_address
			if adr.ip?
				cl = TCPSocket.new(adr.ip_address, adr.ip_port)
				cl.autoclose = false
			else
				cl = UNIXSocket.new(adr.unix_path)
				cl.autoclose = false
			end
		end
		cl.write(cancel_request)
		cl.read(1)
	else
		cl = socket_io.remote_address.connect
		# Send CANCEL_REQUEST_CODE and parameters
		cl.write(cancel_request)
		# Wait for the postmaster to close the connection, which indicates that it's processed the request.
		cl.read(1)
	end
	cl.close
	nil
rescue SystemCallError => err
	err.to_s
end

def conndefaults
	return self.class.conndefaults
end

def conndefaults_hash
	return self.class.conndefaults_hash
end

def connect_internal(opts, errors=nil)
 self.connect_start(opts) or
	raise(PG::Error, "Unable to create a new connection")
PG::ConnectionBad.new(conn.error_message, connection: self) if conn.status == PG::CONNECTION_BAD
end(:async_connect_or_reset, :connect_poll)
PG::ConnectionBad => err
ors && !(conn && [PG::CONNECTION_AWAITING_RESPONSE].include?(conn.instance_variable_get(:@last_status)))
ms to be no authentication error -> try next host
s << err
n nil
bably an authentication error

def connect_to_hosts(*args)
string = parse_connect_args(*args)
 PG::Connection.conninfo_parse(option_string).each_with_object({}){|h, o| o[h[:keyword].to_sym] = h[:val] if h[:val] }
 PG::Connection.conndefaults.each_with_object({}){|h, o| o[h[:keyword].to_sym] = h[:val] if h[:val] }.merge(iopts)
= []
s[:hostaddr]
addr is provided -> no need to resolve hostnames
ddrs = iopts[:hostaddr].split(",", -1)
 = iopts[:host].split(",", -1) if iopts[:host]
PG::ConnectionBad, "could not match #{ihosts.size} host names to #{ihostaddrs.size} hostaddr values" if ihosts && ihosts.size != ihostaddrs.size
 = iopts[:port].split(",", -1)
 = iports * ihostaddrs.size if iports.size == 1
PG::ConnectionBad, "could not match #{iports.size} port numbers to #{ihostaddrs.size} hosts" if iports.size != ihostaddrs.size
to connect to each hostaddr with separate timeout
ddrs.each_with_index do |ihostaddr, idx|
 = iopts.merge(hostaddr: ihostaddr, port: iports[idx])
[:host] = ihosts[idx] if ihosts
onnect_internal(oopts, errors)
n c if c
opts[:host] && !iopts[:host].empty?
lve DNS in Ruby to avoid blocking state while connecting, when it ...
 = iopts[:host].split(",", -1)
 = iopts[:port].split(",", -1)
 = iports * ihosts.size if iports.size == 1
PG::ConnectionBad, "could not match #{iports.size} port numbers to #{ihosts.size} hosts" if iports.size != ihosts.size
.each_with_index do |mhost, idx|
s host_is_named_pipe?(mhost)
s = if Fiber.respond_to?(:scheduler) &&
iber.scheduler &&
UBY_VERSION < '3.1.'
se a second thread to avoid blocking of the scheduler.
TCPSocket.gethostbyname` isn't fiber aware before ruby-3.1.
ead.new{ Addrinfo.getaddrinfo(mhost, nil, nil, :STREAM).map(&:ip_address) rescue [''] }.value

rinfo.getaddrinfo(mhost, nil, nil, :STREAM).map(&:ip_address) rescue ['']
y to connect to each host with separate timeout
s.each do |addr|
ts = iopts.merge(hostaddr: addr, host: mhost, port: iports[idx])
 connect_internal(oopts, errors)
urn c if c
 hostname to resolve (UnixSocket)
s = iopts.merge(host: mhost, port: iports[idx])
connect_internal(oopts, errors)
rn c if c
ost given
 connect_internal(iopts)
G::ConnectionBad, errors.join("\n")

def conninfo_hash
	return self.conninfo.each_with_object({}) do |info, hash|
		hash[ info[:keyword].to_sym ] = info[:val]
	end
end

def copy_data( sql, coder=nil )
	raise PG::NotInBlockingMode.new("copy_data can not be used in nonblocking mode", connection: self) if nonblocking?
	res = exec( sql )
	case res.result_status
	when PGRES_COPY_IN
		begin
			if coder
				old_coder = self.encoder_for_put_copy_data
				self.encoder_for_put_copy_data = coder
			end
			yield res
		rescue Exception => err
			errmsg = "%s while copy data: %s" % [ err.class.name, err.message ]
			put_copy_end( errmsg )
			get_result
			raise
		else
			put_copy_end
			get_last_result
		ensure
			self.encoder_for_put_copy_data = old_coder if coder
		end
	when PGRES_COPY_OUT
		begin
			if coder
				old_coder = self.decoder_for_get_copy_data
				self.decoder_for_get_copy_data = coder
			end
			yield res
		rescue Exception => err
			cancel
			begin
				while get_copy_data
				end
			rescue PG::Error
				# Ignore error in cleanup to avoid losing original exception
			end
			while get_result
			end
			raise err
		else
			res = get_last_result
			if !res || res.result_status != PGRES_COMMAND_OK
				while get_copy_data
				end
				while get_result
				end
				raise PG::NotAllCopyDataRetrieved.new("Not all COPY data retrieved", connection: self)
			end
			res
		ensure
			self.decoder_for_get_copy_data = old_coder if coder
		end
	else
		raise ArgumentError, "SQL command is no COPY statement: #{sql}"
	end
end

def encrypt_password( password, username, algorithm=nil )
	algorithm ||= exec("SHOW password_encryption").getvalue(0,0)
	sync_encrypt_password(password, username, algorithm)
end

def get_copy_data(async=false, decoder=nil)
	if async
		return sync_get_copy_data(async, decoder)
	else
		while (res=sync_get_copy_data(true, decoder)) == false
			socket_io.wait_readable
			consume_input
		end
		return res
	end
end

def get_result
	block
	sync_get_result
end

def host_is_named_pipe?(host_string)
ring.empty? || host_string.start_with?("/") ||  # it's UnixSocket?
t_string.start_with?("@") ||  # it's UnixSocket in the abstract namespace?
t's a path on Windows?
BY_PLATFORM =~ /mingw|mswin/ && host_string =~ /\A([\/\\]|\w:[\/\\])/)

def isnonblocking
	false
end

def new(*args)
	conn = connect_to_hosts(*args)
	if block_given?
		begin
			return yield conn
		ensure
			conn.finish
		end
	end
	conn
end

def ping(*args)
	if Fiber.respond_to?(:scheduler) && Fiber.scheduler
		# Run PQping in a second thread to avoid blocking of the scheduler.
		# Unfortunately there's no nonblocking way to run ping.
		Thread.new { sync_ping(*args) }.value
	else
		sync_ping(*args)
	end
end

def put_copy_data(buffer, encoder=nil)
	# sync_put_copy_data does a non-blocking attept to flush data.
	until res=sync_put_copy_data(buffer, encoder)
		# It didn't flush immediately and allocation of more buffering memory failed.
		# Wait for all data sent by doing a blocking flush.
		res = flush
	end
	# And do a blocking flush every 100 calls.
	# This is to avoid memory bloat, when sending the data is slower than calls to put_copy_data happen.
	if (@calls_to_put_copy_data += 1) > 100
		@calls_to_put_copy_data = 0
		res = flush
	end
	res
end

def put_copy_end(*args)
	until sync_put_copy_end(*args)
		flush
	end
	@calls_to_put_copy_data = 0
	flush
end

def reset
	reset_start
	async_connect_or_reset(:reset_poll)
	self
end

def setnonblocking(enabled)
	singleton_class.async_send_api = !enabled
	self.flush_data = !enabled
	sync_setnonblocking(true)
end

def ssl_attributes
	ssl_attribute_names.each.with_object({}) do |n,h|
		h[n] = ssl_attribute(n)
	end
end

def transaction
	rollback = false
	exec "BEGIN"
	yield(self)
rescue Exception
	rollback = true
	cancel if transaction_status == PG::PQTRANS_ACTIVE
	block
	exec "ROLLBACK"
	raise
ensure
	exec "COMMIT" unless rollback
end