Class: PG::Connection

# File 'lib/pg/connection.rb', line 923

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

# File 'lib/pg/connection.rb', line 876

alias async_ping ping

# File 'ext/pg_connection.c', line 377

static VALUE
pgconn_s_conndefaults(VALUE self)
{
	PQconninfoOption *options = PQconndefaults();
	VALUE array = pgconn_make_conninfo_array( options );

	PQconninfoFree(options);

	UNUSED( self );

	return array;
}

# File 'lib/pg/connection.rb', line 329

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

# File 'lib/pg/connection.rb', line 776

alias connect new

# File 'lib/pg/connection.rb', line 44

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

# File 'ext/pg_connection.c', line 314

static VALUE
pgconn_s_connect_start( int argc, VALUE *argv, VALUE klass )
{
	VALUE rb_conn;
	VALUE conninfo;
	t_pg_connection *this;

	/*
	 * PG::Connection.connect_start must act as both alloc() and initialize()
	 * because it is not invoked by calling new().
	 */
	rb_conn  = pgconn_s_allocate( klass );
	this = pg_get_connection( rb_conn );
	conninfo = rb_funcall2( klass, rb_intern("parse_connect_args"), argc, argv );
	this->pgconn = gvl_PQconnectStart( StringValueCStr(conninfo) );

	if( this->pgconn == NULL )
		rb_raise(rb_ePGerror, "PQconnectStart() unable to allocate PGconn structure");

	if ( PQstatus(this->pgconn) == CONNECTION_BAD )
		pg_raise_conn_error( rb_eConnectionBad, rb_conn, "%s", PQerrorMessage(this->pgconn));

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_conn, pgconn_finish, rb_conn );
	}
	return rb_conn;
}

# File 'lib/pg/connection.rb', line 815

private def connect_to_hosts(*args)
	option_string = parse_connect_args(*args)
	iopts = PG::Connection.conninfo_parse(option_string).each_with_object({}){|h, o| o[h[:keyword].to_sym] = h[:val] if h[:val] }
	iopts = PG::Connection.conndefaults.each_with_object({}){|h, o| o[h[:keyword].to_sym] = h[:val] if h[:val] }.merge(iopts)

	if iopts[:hostaddr]
		# hostaddr is provided -> no need to resolve hostnames

	elsif iopts[:host] && !iopts[:host].empty? && PG.library_version >= 100000
		iopts = resolve_hosts(iopts)
	else
		# No host given
	end
	conn = self.connect_start(iopts) or
								raise(PG::Error, "Unable to create a new connection")

	raise PG::ConnectionBad, conn.error_message if conn.status == PG::CONNECTION_BAD

	conn.send(:async_connect_or_reset, :connect_poll)
	conn
end

# File 'ext/pg_connection.c', line 400

static VALUE
pgconn_s_conninfo_parse(VALUE self, VALUE conninfo)
{
	VALUE array;
	char *errmsg = NULL;
	PQconninfoOption *options = PQconninfoParse(StringValueCStr(conninfo), &errmsg);
	if(errmsg){
		VALUE error = rb_str_new_cstr(errmsg);
		PQfreemem(errmsg);
		rb_raise(rb_ePGerror, "%"PRIsVALUE, error);
	}
	array = pgconn_make_conninfo_array( options );

	PQconninfoFree(options);

	UNUSED( self );

	return array;
}

# File 'ext/pg_connection.c', line 456

static VALUE
pgconn_s_encrypt_password(VALUE self, VALUE password, VALUE username)
{
	char *encrypted = NULL;
	VALUE rval = Qnil;

	UNUSED( self );

	Check_Type(password, T_STRING);
	Check_Type(username, T_STRING);

	encrypted = PQencryptPassword(StringValueCStr(password), StringValueCStr(username));
	rval = rb_str_new2( encrypted );
	PQfreemem( encrypted );

	return rval;
}

# File 'ext/pg_connection.c', line 1627

static VALUE
pgconn_s_escape(VALUE self, VALUE string)
{
	size_t size;
	int error;
	VALUE result;
	int enc_idx;
	int singleton = !rb_obj_is_kind_of(self, rb_cPGconn);

	StringValueCStr(string);
	enc_idx = singleton ? ENCODING_GET(string) : pg_get_connection(self)->enc_idx;
	if( ENCODING_GET(string) != enc_idx ){
		string = rb_str_export_to_enc(string, rb_enc_from_index(enc_idx));
	}

	result = rb_str_new(NULL, RSTRING_LEN(string) * 2 + 1);
	PG_ENCODING_SET_NOCHECK(result, enc_idx);
	if( !singleton ) {
		size = PQescapeStringConn(pg_get_pgconn(self), RSTRING_PTR(result),
			RSTRING_PTR(string), RSTRING_LEN(string), &error);
		if(error)
			pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(pg_get_pgconn(self)));

	} else {
		size = PQescapeString(RSTRING_PTR(result), RSTRING_PTR(string), RSTRING_LEN(string));
	}
	rb_str_set_len(result, size);

	return result;
}

# File 'ext/pg_connection.c', line 1681

static VALUE
pgconn_s_escape_bytea(VALUE self, VALUE str)
{
	unsigned char *from, *to;
	size_t from_len, to_len;
	VALUE ret;

	Check_Type(str, T_STRING);
	from      = (unsigned char*)RSTRING_PTR(str);
	from_len  = RSTRING_LEN(str);

	if ( rb_obj_is_kind_of(self, rb_cPGconn) ) {
		to = PQescapeByteaConn(pg_get_pgconn(self), from, from_len, &to_len);
	} else {
		to = PQescapeBytea( from, from_len, &to_len);
	}

	ret = rb_str_new((char*)to, to_len - 1);
	PQfreemem(to);
	return ret;
}

# File 'lib/pg/connection.rb', line 837

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

# File 'lib/pg/connection.rb', line 777

alias open new

# File 'lib/pg/connection.rb', line 64

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.to_s
		when /=/, /:\/\//
			# 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] = PROGRAM_NAME.sub( /^(.{30}).{4,}(.{30})$/ ){ $1"..."$2 }
	end

	return connect_hash_to_string(iopts)
end

# File 'lib/pg/connection.rb', line 867

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

# File 'lib/pg/connection.rb', line 37

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

# File 'ext/pg_connection.c', line 3013

static VALUE
pgconn_s_quote_ident(VALUE self, VALUE str_or_array)
{
	VALUE ret;
	int enc_idx;

	if( rb_obj_is_kind_of(self, rb_cPGconn) ){
		enc_idx = pg_get_connection(self)->enc_idx;
	}else{
		enc_idx = RB_TYPE_P(str_or_array, T_STRING) ? ENCODING_GET( str_or_array ) : rb_ascii8bit_encindex();
	}
	pg_text_enc_identifier(NULL, str_or_array, NULL, &ret, enc_idx);

	return ret;
}

# File 'lib/pg/connection.rb', line 784

private def resolve_hosts(iopts)
	ihosts = iopts[:host].split(",", -1)
	iports = iopts[:port].split(",", -1)
	iports = [nil] if iports.size == 0
	iports = iports * ihosts.size if iports.size == 1
	raise PG::ConnectionBad, "could not match #{iports.size} port numbers to #{ihosts.size} hosts" if iports.size != ihosts.size

	dests = ihosts.each_with_index.flat_map do |mhost, idx|
		unless host_is_named_pipe?(mhost)
			if Fiber.respond_to?(:scheduler) &&
						Fiber.scheduler &&
						RUBY_VERSION < '3.1.'

				# Use a second thread to avoid blocking of the scheduler.
				# `TCPSocket.gethostbyname` isn't fiber aware before ruby-3.1.
				hostaddrs = Thread.new{ Addrinfo.getaddrinfo(mhost, nil, nil, :STREAM).map(&:ip_address) rescue [''] }.value
			else
				hostaddrs = Addrinfo.getaddrinfo(mhost, nil, nil, :STREAM).map(&:ip_address) rescue ['']
			end
		else
			# No hostname to resolve (UnixSocket)
			hostaddrs = [nil]
		end
		hostaddrs.map { |hostaddr| [hostaddr, mhost, iports[idx]] }
	end
	iopts.merge(
		hostaddr: dests.map{|d| d[0] }.join(","),
		host: dests.map{|d| d[1] }.join(","),
		port: dests.map{|d| d[2] }.join(","))
end

# File 'lib/pg/connection.rb', line 778

alias setdb new

# File 'lib/pg/connection.rb', line 779

alias setdblogin new

# File 'ext/pg_connection.c', line 271

static VALUE
pgconn_s_sync_connect(int argc, VALUE *argv, VALUE klass)
{
	t_pg_connection *this;
	VALUE conninfo;
	VALUE self = pgconn_s_allocate( klass );

	this = pg_get_connection( self );
	conninfo = rb_funcall2( rb_cPGconn, rb_intern("parse_connect_args"), argc, argv );
	this->pgconn = gvl_PQconnectdb(StringValueCStr(conninfo));

	if(this->pgconn == NULL)
		rb_raise(rb_ePGerror, "PQconnectdb() unable to allocate PGconn structure");

	if (PQstatus(this->pgconn) == CONNECTION_BAD)
		pg_raise_conn_error( rb_eConnectionBad, self, "%s", PQerrorMessage(this->pgconn));

	pgconn_set_default_encoding( self );

	if (rb_block_given_p()) {
		return rb_ensure(rb_yield, self, pgconn_finish, self);
	}
	return self;
}

# File 'ext/pg_connection.c', line 342

static VALUE
pgconn_s_sync_ping( int argc, VALUE *argv, VALUE klass )
{
	PGPing ping;
	VALUE conninfo;

	conninfo = rb_funcall2( klass, rb_intern("parse_connect_args"), argc, argv );
	ping     = gvl_PQping( StringValueCStr(conninfo) );

	return INT2FIX((int)ping);
}

# File 'ext/pg_connection.c', line 1713

static VALUE
pgconn_s_unescape_bytea(VALUE self, VALUE str)
{
	unsigned char *from, *to;
	size_t to_len;
	VALUE ret;

	UNUSED( self );

	Check_Type(str, T_STRING);
	from = (unsigned char*)StringValueCStr(str);

	to = PQunescapeBytea(from, &to_len);

	ret = rb_str_new((char*)to, to_len);
	PQfreemem(to);
	return ret;
}

# File 'ext/pg_connection.c', line 4367

static VALUE
pgconn_decoder_for_get_copy_data_set(VALUE self, VALUE decoder)
{
	t_pg_connection *this = pg_get_connection( self );

	rb_check_frozen(self);
	if( decoder != Qnil ){
		t_pg_coder *co;
		UNUSED(co);
		/* Check argument type */
		TypedData_Get_Struct(decoder, t_pg_coder, &pg_coder_type, co);
	}
	RB_OBJ_WRITE(self, &this->decoder_for_get_copy_data, decoder);

	return decoder;
}

# File 'ext/pg_connection.c', line 4347

static VALUE
pgconn_encoder_for_put_copy_data_get(VALUE self)
{
	t_pg_connection *this = pg_get_connection( self );

	return this->encoder_for_put_copy_data;
}

# File 'ext/pg_connection.c', line 4318

static VALUE
pgconn_encoder_for_put_copy_data_set(VALUE self, VALUE encoder)
{
	t_pg_connection *this = pg_get_connection( self );

	rb_check_frozen(self);
	if( encoder != Qnil ){
		t_pg_coder *co;
		UNUSED(co);
		/* Check argument type */
		TypedData_Get_Struct(encoder, t_pg_coder, &pg_coder_type, co);
	}
	RB_OBJ_WRITE(self, &this->encoder_for_put_copy_data, encoder);

	return encoder;
}

# File 'ext/pg_connection.c', line 4443

static VALUE
pgconn_field_name_type_get(VALUE self)
{
	t_pg_connection *this = pg_get_connection( self );

	if( this->flags & PG_RESULT_FIELD_NAMES_SYMBOL ){
		return sym_symbol;
	} else if( this->flags & PG_RESULT_FIELD_NAMES_STATIC_SYMBOL ){
		return sym_static_symbol;
	} else {
		return sym_string;
	}
}

# File 'ext/pg_connection.c', line 4420

static VALUE
pgconn_field_name_type_set(VALUE self, VALUE sym)
{
	t_pg_connection *this = pg_get_connection( self );

	rb_check_frozen(self);
	this->flags &= ~PG_RESULT_FIELD_NAMES_MASK;
	if( sym == sym_symbol ) this->flags |= PG_RESULT_FIELD_NAMES_SYMBOL;
	else if ( sym == sym_static_symbol ) this->flags |= PG_RESULT_FIELD_NAMES_STATIC_SYMBOL;
	else if ( sym == sym_string );
	else rb_raise(rb_eArgError, "invalid argument %+"PRIsVALUE, sym);

	return sym;
}

# File 'ext/pg_connection.c', line 549

static VALUE
pgconn_finished_p( VALUE self )
{
	t_pg_connection *this = pg_get_connection( self );
	if ( this->pgconn ) return Qfalse;
	return Qtrue;
}

# File 'ext/pg_connection.c', line 2495

static VALUE
pgconn_flush_data_set( VALUE self, VALUE enabled ){
	t_pg_connection *conn = pg_get_connection(self);
	rb_check_frozen(self);
	conn->flush_data = RTEST(enabled);
	return enabled;
}

# File 'ext/pg_connection.c', line 4067

static VALUE
pgconn_internal_encoding(VALUE self)
{
	PGconn *conn = pg_get_pgconn( self );
	rb_encoding *enc = pg_conn_enc_get( conn );

	if ( enc ) {
		return rb_enc_from_encoding( enc );
	} else {
		return Qnil;
	}
}

# File 'ext/pg_connection.c', line 4094

static VALUE
pgconn_internal_encoding_set(VALUE self, VALUE enc)
{
	rb_check_frozen(self);
	if (NIL_P(enc)) {
		pgconn_sync_set_client_encoding( self, rb_usascii_str_new_cstr("SQL_ASCII") );
		return enc;
	}
	else if ( TYPE(enc) == T_STRING && strcasecmp("JOHAB", StringValueCStr(enc)) == 0 ) {
		pgconn_sync_set_client_encoding(self, rb_usascii_str_new_cstr("JOHAB"));
		return enc;
	}
	else {
		rb_encoding *rbenc = rb_to_encoding( enc );
		const char *name = pg_get_rb_encoding_as_pg_encoding( rbenc );

		if ( gvl_PQsetClientEncoding(pg_get_pgconn( self ), name) == -1 ) {
			VALUE server_encoding = pgconn_external_encoding( self );
			rb_raise( rb_eEncCompatError, "incompatible character encodings: %s and %s",
					  rb_enc_name(rb_to_encoding(server_encoding)), name );
		}
		pgconn_set_internal_encoding_index( self );
		return enc;
	}
}

# File 'lib/pg/connection.rb', line 475

alias nonblocking? isnonblocking

# File 'ext/pg_connection.c', line 3517

static VALUE
pgconn_ssl_in_use(VALUE self)
{
	return PQsslInUse(pg_get_pgconn(self)) ? Qtrue : Qfalse;
}

# File 'ext/pg_connection.c', line 4259

static VALUE
pgconn_type_map_for_queries_get(VALUE self)
{
	t_pg_connection *this = pg_get_connection( self );

	return this->type_map_for_queries;
}

# File 'ext/pg_connection.c', line 4235

static VALUE
pgconn_type_map_for_queries_set(VALUE self, VALUE typemap)
{
	t_pg_connection *this = pg_get_connection( self );
	t_typemap *tm;
	UNUSED(tm);

	rb_check_frozen(self);
	/* Check type of method param */
	TypedData_Get_Struct(typemap, t_typemap, &pg_typemap_type, tm);

	RB_OBJ_WRITE(self, &this->type_map_for_queries, typemap);

	return typemap;
}

# File 'ext/pg_connection.c', line 4297

static VALUE
pgconn_type_map_for_results_get(VALUE self)
{
	t_pg_connection *this = pg_get_connection( self );

	return this->type_map_for_results;
}

# File 'ext/pg_connection.c', line 4276

static VALUE
pgconn_type_map_for_results_set(VALUE self, VALUE typemap)
{
	t_pg_connection *this = pg_get_connection( self );
	t_typemap *tm;
	UNUSED(tm);

	rb_check_frozen(self);
	TypedData_Get_Struct(typemap, t_typemap, &pg_typemap_type, tm);
	RB_OBJ_WRITE(self, &this->type_map_for_results, typemap);

	return typemap;
}

# File 'lib/pg/connection.rb', line 642

private def async_connect_or_reset(poll_meth)
	# Track the progress of the connection, waiting for the socket to become readable/writable before polling it

	if (timeo = conninfo_hash[:connect_timeout].to_i) && timeo > 0
		# Lowest timeout is 2 seconds - like in libpq
		timeo = [timeo, 2].max
		host_count = conninfo_hash[:host].to_s.count(",") + 1
		stop_time = timeo * host_count + Process.clock_gettime(Process::CLOCK_MONOTONIC)
	end

	poll_status = PG::PGRES_POLLING_WRITING
	until poll_status == PG::PGRES_POLLING_OK ||
			poll_status == PG::PGRES_POLLING_FAILED

		# Set single timeout to parameter "connect_timeout" but
		# don't exceed total connection time of number-of-hosts * connect_timeout.
		timeout = [timeo, stop_time - Process.clock_gettime(Process::CLOCK_MONOTONIC)].min if stop_time
		event = if !timeout || timeout >= 0
			# If the socket needs to read, wait 'til it becomes readable to poll again
			case poll_status
			when PG::PGRES_POLLING_READING
				if defined?(IO::READABLE) # ruby-3.0+
					socket_io.wait(IO::READABLE | IO::PRIORITY, timeout)
				else
					IO.select([socket_io], nil, [socket_io], timeout)
				end

			# ...and the same for when the socket needs to write
			when PG::PGRES_POLLING_WRITING
				if defined?(IO::WRITABLE) # ruby-3.0+
					# Use wait instead of wait_readable, since connection errors are delivered as
					# exceptional/priority events on Windows.
					socket_io.wait(IO::WRITABLE | IO::PRIORITY, timeout)
				else
					# io#wait on ruby-2.x doesn't wait for priority, so fallback to IO.select
					IO.select(nil, [socket_io], [socket_io], timeout)
				end
			end
		end
		# connection to server at "localhost" (127.0.0.1), port 5433 failed: timeout expired (PG::ConnectionBad)
		# connection to server on socket "/var/run/postgresql/.s.PGSQL.5433" failed: No such file or directory
		unless event
			if self.class.send(:host_is_named_pipe?, host)
				connhost = "on socket \"#{host}\""
			elsif respond_to?(:hostaddr)
				connhost = "at \"#{host}\" (#{hostaddr}), port #{port}"
			else
				connhost = "at \"#{host}\", port #{port}"
			end
			raise PG::ConnectionBad.new("connection to server #{connhost} failed: timeout expired", connection: self)
		end

		# Check to see if it's finished or failed yet
		poll_status = send( poll_meth )
	end

	unless status == PG::CONNECTION_OK
		msg = error_message
		finish
		raise PG::ConnectionBad.new(msg, connection: self)
	end

	# Set connection to nonblocking to handle all blocking states in ruby.
	# That way a fiber scheduler is able to handle IO requests.
	sync_setnonblocking(true)
	self.flush_data = true
	set_default_encoding
end

# File 'lib/pg/connection.rb', line 559

alias async_encrypt_password encrypt_password

# File 'lib/pg/connection.rb', line 434

alias async_get_copy_data get_copy_data

# File 'lib/pg/connection.rb', line 404

alias async_get_result get_result

# File 'lib/pg/connection.rb', line 474

alias async_isnonblocking isnonblocking

# File 'lib/pg/connection.rb', line 511

alias async_put_copy_data put_copy_data

# File 'lib/pg/connection.rb', line 532

alias async_put_copy_end put_copy_end

# File 'lib/pg/connection.rb', line 577

alias async_reset reset

# File 'lib/pg/connection.rb', line 462

alias async_setnonblocking setnonblocking

# File 'ext/pg_connection.c', line 1010

static VALUE
pgconn_backend_key(VALUE self)
{
	int be_key;
	struct pg_cancel *cancel;
	PGconn *conn = pg_get_pgconn(self);

	cancel = (struct pg_cancel*)PQgetCancel(conn);
	if(cancel == NULL)
		pg_raise_conn_error( rb_ePGerror, self, "Invalid connection!");

	if( cancel->be_pid != PQbackendPID(conn) )
		rb_raise(rb_ePGerror,"Unexpected binary struct layout - please file a bug report at ruby-pg!");

	be_key = cancel->be_key;

	PQfreeCancel(cancel);

	return INT2NUM(be_key);
}

# File 'ext/pg_connection.c', line 980

static VALUE
pgconn_backend_pid(VALUE self)
{
	return INT2NUM(PQbackendPID(pg_get_pgconn(self)));
}

# File 'ext/pg_connection.c', line 3050

VALUE
pgconn_block( int argc, VALUE *argv, VALUE self ) {
	struct timeval timeout;
	struct timeval *ptimeout = NULL;
	VALUE timeout_in;
	double timeout_sec;
	void *ret;

	if ( rb_scan_args(argc, argv, "01", &timeout_in) == 1 ) {
		timeout_sec = NUM2DBL( timeout_in );
		timeout.tv_sec = (time_t)timeout_sec;
		timeout.tv_usec = (suseconds_t)((timeout_sec - (long)timeout_sec) * 1e6);
		ptimeout = &timeout;
	}

	ret = wait_socket_readable( self, ptimeout, get_result_readable);

	if( !ret )
		return Qfalse;

	return Qtrue;
}

# File 'lib/pg/connection.rb', line 587

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

# File 'lib/pg/connection.rb', line 379

def check_socket
	while socket_io.wait_readable(0)
		consume_input
	end
	nil
end

# File 'ext/pg_connection.c', line 4147

static VALUE
pgconn_async_set_client_encoding(VALUE self, VALUE encname)
{
	VALUE query_format, query;

	rb_check_frozen(self);
	Check_Type(encname, T_STRING);
	query_format = rb_str_new_cstr("set client_encoding to '%s'");
	query = rb_funcall(query_format, rb_intern("%"), 1, encname);

	pgconn_async_exec(1, &query, self);
	pgconn_set_internal_encoding_index( self );

	return Qnil;
}

# File 'ext/pg_connection.c', line 531

static VALUE
pgconn_finish( VALUE self )
{
	t_pg_connection *this = pg_get_connection_safe( self );

	pgconn_close_socket_io( self );
	PQfinish( this->pgconn );
	this->pgconn = NULL;
	return Qnil;
}

# File 'lib/pg/connection.rb', line 321

def conndefaults
	return self.class.conndefaults
end

# File 'lib/pg/connection.rb', line 337

def conndefaults_hash
	return self.class.conndefaults_hash
end

# File 'ext/pg_connection.c', line 514

static VALUE
pgconn_connect_poll(VALUE self)
{
	PostgresPollingStatusType status;

	pgconn_close_socket_io(self);
	status = gvl_PQconnectPoll(pg_get_pgconn(self));

	return INT2FIX((int)status);
}

# File 'ext/pg_connection.c', line 1038

static VALUE
pgconn_connection_needs_password(VALUE self)
{
	return PQconnectionNeedsPassword(pg_get_pgconn(self)) ? Qtrue : Qfalse;
}

# File 'ext/pg_connection.c', line 1051

static VALUE
pgconn_connection_used_password(VALUE self)
{
	return PQconnectionUsedPassword(pg_get_pgconn(self)) ? Qtrue : Qfalse;
}

# File 'ext/pg_connection.c', line 764

static VALUE
pgconn_conninfo( VALUE self )
{
	PGconn *conn = pg_get_pgconn(self);
	PQconninfoOption *options = PQconninfo( conn );
	VALUE array = pgconn_make_conninfo_array( options );

	PQconninfoFree(options);

	return array;
}

# File 'lib/pg/connection.rb', line 345

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

# File 'ext/pg_connection.c', line 2143

static VALUE
pgconn_consume_input(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	/* returns 0 on error */
	if(PQconsumeInput(conn) == 0) {
		pgconn_close_socket_io(self);
		pg_raise_conn_error( rb_eConnectionBad, self, "%s", PQerrorMessage(conn));
	}

	return Qnil;
}

# File 'lib/pg/connection.rb', line 211

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 && res.binary_tuples == 1
				# Binary file header (11 byte signature, 32 bit flags and 32 bit extension length)
				put_copy_data(BinarySignature + ("\x00" * 8))
			end

			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 ]
			begin
				put_copy_end( errmsg )
			rescue PG::Error
				# Ignore error in cleanup to avoid losing original exception
			end
			discard_results
			raise err
		else
			begin
				self.encoder_for_put_copy_data = old_coder if coder

				if coder && res.binary_tuples == 1
					put_copy_data("\xFF\xFF") # Binary file trailer 16 bit "-1"
				end

				put_copy_end
			rescue PG::Error => err
				raise PG::LostCopyState.new("#{err} (probably by executing another SQL query while running a COPY command)", connection: self)
			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
			cancel
			discard_results
			raise
		else
			if coder && res.binary_tuples == 1
				# There are two end markers in binary mode: file trailer and the final nil.
				# The file trailer is expected to be processed by BinaryDecoder::CopyRow and already returns nil, so that the remaining NULL from PQgetCopyData is retrieved here:
				if get_copy_data
					discard_results
					raise PG::NotAllCopyDataRetrieved.new("Not all binary COPY data retrieved", connection: self)
				end
			end
			res = get_last_result
			if !res
				discard_results
				raise PG::LostCopyState.new("Lost COPY state (probably by executing another SQL query while running a COPY command)", connection: self)
			elsif res.result_status != PGRES_COMMAND_OK
				discard_results
				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

# File 'ext/pg_connection.c', line 632

static VALUE
pgconn_db(VALUE self)
{
	char *db = PQdb(pg_get_pgconn(self));
	if (!db) return Qnil;
	return rb_str_new2(db);
}

# File 'ext/pg_connection.c', line 3468

static VALUE
pgconn_async_describe_portal(VALUE self, VALUE portal)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	pgconn_send_describe_portal( self, portal );
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 3492

static VALUE
pgconn_async_describe_prepared(VALUE self, VALUE stmt_name)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	pgconn_send_describe_prepared( self, stmt_name );
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 3175

static VALUE
pgconn_discard_results(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	VALUE socket_io;

	switch( PQtransactionStatus(conn) ) {
		case PQTRANS_IDLE:
		case PQTRANS_INTRANS:
		case PQTRANS_INERROR:
			return Qnil;
		default:;
	}

	socket_io = pgconn_socket_io(self);

	for(;;) {
		PGresult *cur;
		int status;

		/* pgconn_block() raises an exception in case of errors.
		* To avoid this call pg_rb_io_wait() and PQconsumeInput() without rb_raise().
		*/
		while( gvl_PQisBusy(conn) ){
			int events;

			switch( PQflush(conn) ) {
				case 1:
					events = RB_NUM2INT(pg_rb_io_wait(socket_io, RB_INT2NUM(PG_RUBY_IO_READABLE | PG_RUBY_IO_WRITABLE), Qnil));
					if (events & PG_RUBY_IO_READABLE){
						if ( PQconsumeInput(conn) == 0 ) goto error;
					}
					break;
				case 0:
					pg_rb_io_wait(socket_io, RB_INT2NUM(PG_RUBY_IO_READABLE), Qnil);
					if ( PQconsumeInput(conn) == 0 ) goto error;
					break;
				default:
					goto error;
			}
		}

		cur = gvl_PQgetResult(conn);
		if( cur == NULL) break;

		status = PQresultStatus(cur);
		PQclear(cur);
		if (status == PGRES_COPY_IN){
			while( gvl_PQputCopyEnd(conn, "COPY terminated by new query or discard_results") == 0 ){
				pgconn_async_flush(self);
			}
		}
		if (status == PGRES_COPY_OUT){
			for(;;) {
				char *buffer = NULL;
				int st = gvl_PQgetCopyData(conn, &buffer, 1);
				if( st == 0 ) {
					/* would block -> wait for readable data */
					pg_rb_io_wait(socket_io, RB_INT2NUM(PG_RUBY_IO_READABLE), Qnil);
					if ( PQconsumeInput(conn) == 0 ) goto error;
				} else if( st > 0 ) {
					/* some data retrieved -> discard it */
					PQfreemem(buffer);
				} else {
					/* no more data */
					break;
				}
			}
		}
	}

	return Qtrue;

error:
	pgconn_close_socket_io(self);
	return Qfalse;
}

# File 'lib/pg/connection.rb', line 555

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

# File 'ext/pg_connection.c', line 3622

static VALUE
pgconn_enter_pipeline_mode(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	int res = PQenterPipelineMode(conn);
	if( res != 1 )
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));

	return Qnil;
}

# File 'ext/pg_connection.c', line 885

static VALUE
pgconn_error_message(VALUE self)
{
	char *error = PQerrorMessage(pg_get_pgconn(self));
	if (!error) return Qnil;
	return rb_str_new2(error);
}

# File 'ext/pg_connection.c', line 1774

static VALUE
pgconn_escape_identifier(VALUE self, VALUE string)
{
	t_pg_connection *this = pg_get_connection_safe( self );
	char *escaped = NULL;
	VALUE result = Qnil;
	int enc_idx = this->enc_idx;

	StringValueCStr(string);
	if( ENCODING_GET(string) != enc_idx ){
		string = rb_str_export_to_enc(string, rb_enc_from_index(enc_idx));
	}

	escaped = PQescapeIdentifier(this->pgconn, RSTRING_PTR(string), RSTRING_LEN(string));
	if (escaped == NULL)
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(this->pgconn));

	result = rb_str_new2(escaped);
	PQfreemem(escaped);
	PG_ENCODING_SET_NOCHECK(result, enc_idx);

	return result;
}

# File 'ext/pg_connection.c', line 1740

static VALUE
pgconn_escape_literal(VALUE self, VALUE string)
{
	t_pg_connection *this = pg_get_connection_safe( self );
	char *escaped = NULL;
	VALUE result = Qnil;
	int enc_idx = this->enc_idx;

	StringValueCStr(string);
	if( ENCODING_GET(string) != enc_idx ){
		string = rb_str_export_to_enc(string, rb_enc_from_index(enc_idx));
	}

	escaped = PQescapeLiteral(this->pgconn, RSTRING_PTR(string), RSTRING_LEN(string));
	if (escaped == NULL)
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(this->pgconn));

	result = rb_str_new2(escaped);
	PQfreemem(escaped);
	PG_ENCODING_SET_NOCHECK(result, enc_idx);

	return result;
}

# File 'ext/pg_connection.c', line 3281

static VALUE
pgconn_async_exec(int argc, VALUE *argv, VALUE self)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	pgconn_send_query( argc, argv, self );
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 3347

static VALUE
pgconn_async_exec_params(int argc, VALUE *argv, VALUE self)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	/* If called with no or nil parameters, use PQsendQuery for compatibility */
	if ( argc == 1 || (argc >= 2 && argc <= 4 && NIL_P(argv[1]) )) {
		pg_deprecated(3, ("forwarding async_exec_params to async_exec is deprecated"));
		pgconn_send_query( argc, argv, self );
	} else {
		pgconn_send_query_params( argc, argv, self );
	}
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 3444

static VALUE
pgconn_async_exec_prepared(int argc, VALUE *argv, VALUE self)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	pgconn_send_query_prepared( argc, argv, self );
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 3644

static VALUE
pgconn_exit_pipeline_mode(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	int res = PQexitPipelineMode(conn);
	if( res != 1 )
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));

	return Qnil;
}

# File 'ext/pg_connection.c', line 4129

static VALUE
pgconn_external_encoding(VALUE self)
{
	t_pg_connection *this = pg_get_connection_safe( self );
	rb_encoding *enc = NULL;
	const char *pg_encname = NULL;

	pg_encname = PQparameterStatus( this->pgconn, "server_encoding" );
	enc = pg_get_pg_encname_as_rb_encoding( pg_encname );
	return rb_enc_from_encoding( enc );
}

# File 'ext/pg_connection.c', line 2471

static VALUE
pgconn_async_flush(VALUE self)
{
	while( pgconn_sync_flush(self) == Qfalse ){
		/* wait for the socket to become read- or write-ready */
		int events;
		VALUE socket_io = pgconn_socket_io(self);
		events = RB_NUM2INT(pg_rb_io_wait(socket_io, RB_INT2NUM(PG_RUBY_IO_READABLE | PG_RUBY_IO_WRITABLE), Qnil));

		if (events & PG_RUBY_IO_READABLE){
			pgconn_consume_input(self);
		}
	}
	return Qtrue;
}

# File 'ext/pg_connection.c', line 2944

static VALUE
pgconn_get_client_encoding(VALUE self)
{
	char *encoding = (char *)pg_encoding_to_char(PQclientEncoding(pg_get_pgconn(self)));
	return rb_str_new2(encoding);
}

# File 'lib/pg/connection.rb', line 423

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

# File 'ext/pg_connection.c', line 3127

static VALUE
pgconn_async_get_last_result(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	VALUE rb_pgresult = Qnil;
	PGresult *cur, *prev;

	cur = prev = NULL;
	for(;;) {
		int status;

		/* wait for input (without blocking) before reading each result */
		wait_socket_readable(self, NULL, get_result_readable);

		cur = gvl_PQgetResult(conn);
		if (cur == NULL)
			break;

		if (prev) PQclear(prev);
		prev = cur;

		status = PQresultStatus(cur);
		if (status == PGRES_COPY_OUT || status == PGRES_COPY_IN || status == PGRES_COPY_BOTH)
			break;
	}

	if (prev) {
		rb_pgresult = pg_new_result( prev, self );
		pg_result_check(rb_pgresult);
	}

	return rb_pgresult;
}

# File 'lib/pg/connection.rb', line 400

def get_result
	block
	sync_get_result
end

# File 'ext/pg_connection.c', line 685

static VALUE
pgconn_host(VALUE self)
{
	char *host = PQhost(pg_get_pgconn(self));
	if (!host) return Qnil;
	return rb_str_new2(host);
}

# File 'ext/pg_connection.c', line 704

static VALUE
pgconn_hostaddr(VALUE self)
{
	char *host = PQhostaddr(pg_get_pgconn(self));
	if (!host) return Qnil;
	return rb_str_new2(host);
}

# File 'lib/pg/connection.rb', line 100

def inspect
	str = self.to_s
	str[-1,0] = if finished?
		" finished"
	else
		stats = []
		stats << " status=#{ PG.constants.grep(/CONNECTION_/).find{|c| PG.const_get(c) == status} }" if status != CONNECTION_OK
		stats << " transaction_status=#{ PG.constants.grep(/PQTRANS_/).find{|c| PG.const_get(c) == transaction_status} }" if transaction_status != PG::PQTRANS_IDLE
		stats << " nonblocking=#{ isnonblocking }" if isnonblocking
		stats << " pipeline_status=#{ PG.constants.grep(/PQ_PIPELINE_/).find{|c| PG.const_get(c) == pipeline_status} }" if respond_to?(:pipeline_status) && pipeline_status != PG::PQ_PIPELINE_OFF
		stats << " client_encoding=#{ get_client_encoding }" if get_client_encoding != "UTF8"
		stats << " type_map_for_results=#{ type_map_for_results.to_s }" unless type_map_for_results.is_a?(PG::TypeMapAllStrings)
		stats << " type_map_for_queries=#{ type_map_for_queries.to_s }" unless type_map_for_queries.is_a?(PG::TypeMapAllStrings)
		stats << " encoder_for_put_copy_data=#{ encoder_for_put_copy_data.to_s }" if encoder_for_put_copy_data
		stats << " decoder_for_get_copy_data=#{ decoder_for_get_copy_data.to_s }" if decoder_for_get_copy_data
		" host=#{host} port=#{port} user=#{user}#{stats.join}"
	end
	return str
end

# File 'ext/pg_connection.c', line 2163

static VALUE
pgconn_is_busy(VALUE self)
{
	return gvl_PQisBusy(pg_get_pgconn(self)) ? Qtrue : Qfalse;
}

# File 'lib/pg/connection.rb', line 471

def isnonblocking
	false
end

# File 'ext/pg_connection.c', line 4005

static VALUE
pgconn_loclose(VALUE self, VALUE in_lo_desc)
{
	PGconn *conn = pg_get_pgconn(self);
	int lo_desc = NUM2INT(in_lo_desc);
	int ret;

	BLOCKING_BEGIN(conn)
		ret = lo_close(conn,lo_desc);
	BLOCKING_END(conn)

	if(ret < 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_close failed");

	return Qnil;
}

# File 'ext/pg_connection.c', line 3722

static VALUE
pgconn_locreat(int argc, VALUE *argv, VALUE self)
{
	Oid lo_oid;
	int mode;
	VALUE nmode;
	PGconn *conn = pg_get_pgconn(self);

	if (rb_scan_args(argc, argv, "01", &nmode) == 0)
		mode = INV_READ;
	else
		mode = NUM2INT(nmode);

	BLOCKING_BEGIN(conn)
		lo_oid = lo_creat(conn, mode);
	BLOCKING_END(conn)

	if (lo_oid == 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_creat failed");

	return UINT2NUM(lo_oid);
}

# File 'ext/pg_connection.c', line 3752

static VALUE
pgconn_locreate(VALUE self, VALUE in_lo_oid)
{
	Oid ret, lo_oid;
	PGconn *conn = pg_get_pgconn(self);
	lo_oid = NUM2UINT(in_lo_oid);

	ret = lo_create(conn, lo_oid);
	if (ret == InvalidOid)
		pg_raise_conn_error( rb_ePGerror, self, "lo_create failed");

	return UINT2NUM(ret);
}

# File 'ext/pg_connection.c', line 3799

static VALUE
pgconn_loexport(VALUE self, VALUE lo_oid, VALUE filename)
{
	PGconn *conn = pg_get_pgconn(self);
	Oid oid;
	int ret;
	Check_Type(filename, T_STRING);

	oid = NUM2UINT(lo_oid);

	BLOCKING_BEGIN(conn)
		ret = lo_export(conn, oid, StringValueCStr(filename));
	BLOCKING_END(conn)

	if (ret < 0) {
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));
	}
	return Qnil;
}

# File 'ext/pg_connection.c', line 3774

static VALUE
pgconn_loimport(VALUE self, VALUE filename)
{
	Oid lo_oid;

	PGconn *conn = pg_get_pgconn(self);

	Check_Type(filename, T_STRING);

	BLOCKING_BEGIN(conn)
		lo_oid = lo_import(conn, StringValueCStr(filename));
	BLOCKING_END(conn)

	if (lo_oid == 0) {
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));
	}
	return UINT2NUM(lo_oid);
}

# File 'ext/pg_connection.c', line 3829

static VALUE
pgconn_loopen(int argc, VALUE *argv, VALUE self)
{
	Oid lo_oid;
	int fd, mode;
	VALUE nmode, selfid;
	PGconn *conn = pg_get_pgconn(self);

	rb_scan_args(argc, argv, "11", &selfid, &nmode);
	lo_oid = NUM2UINT(selfid);
	if(NIL_P(nmode))
		mode = INV_READ;
	else
		mode = NUM2INT(nmode);

	BLOCKING_BEGIN(conn)
		fd = lo_open(conn, lo_oid, mode);
	BLOCKING_END(conn)

	if(fd < 0) {
		pg_raise_conn_error( rb_ePGerror, self, "can't open large object: %s", PQerrorMessage(conn));
	}
	return INT2FIX(fd);
}

# File 'ext/pg_connection.c', line 3892

static VALUE
pgconn_loread(VALUE self, VALUE in_lo_desc, VALUE in_len)
{
	int ret;
  PGconn *conn = pg_get_pgconn(self);
	int len = NUM2INT(in_len);
	int lo_desc = NUM2INT(in_lo_desc);
	VALUE str;
	char *buffer;

	if (len < 0)
		pg_raise_conn_error( rb_ePGerror, self, "negative length %d given", len);

	buffer = ALLOC_N(char, len);

	BLOCKING_BEGIN(conn)
		ret = lo_read(conn, lo_desc, buffer, len);
	BLOCKING_END(conn)

	if(ret < 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_read failed");

	if(ret == 0) {
		xfree(buffer);
		return Qnil;
	}

	str = rb_str_new(buffer, ret);
	xfree(buffer);

	return str;
}

# File 'ext/pg_connection.c', line 3934

static VALUE
pgconn_lolseek(VALUE self, VALUE in_lo_desc, VALUE offset, VALUE whence)
{
	PGconn *conn = pg_get_pgconn(self);
	int lo_desc = NUM2INT(in_lo_desc);
	int ret;

	BLOCKING_BEGIN(conn)
		ret = lo_lseek(conn, lo_desc, NUM2INT(offset), NUM2INT(whence));
	BLOCKING_END(conn)

	if(ret < 0) {
		pg_raise_conn_error( rb_ePGerror, self, "lo_lseek failed");
	}

	return INT2FIX(ret);
}

# File 'ext/pg_connection.c', line 3958

static VALUE
pgconn_lotell(VALUE self, VALUE in_lo_desc)
{
	int position;
	PGconn *conn = pg_get_pgconn(self);
	int lo_desc = NUM2INT(in_lo_desc);

	BLOCKING_BEGIN(conn)
		position = lo_tell(conn, lo_desc);
	BLOCKING_END(conn)

	if(position < 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_tell failed");

	return INT2FIX(position);
}

# File 'ext/pg_connection.c', line 3981

static VALUE
pgconn_lotruncate(VALUE self, VALUE in_lo_desc, VALUE in_len)
{
	PGconn *conn = pg_get_pgconn(self);
	int lo_desc = NUM2INT(in_lo_desc);
	size_t len = NUM2INT(in_len);
	int ret;

	BLOCKING_BEGIN(conn)
		ret = lo_truncate(conn,lo_desc,len);
	BLOCKING_END(conn)

	if(ret < 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_truncate failed");

	return Qnil;
}

# File 'ext/pg_connection.c', line 4028

static VALUE
pgconn_lounlink(VALUE self, VALUE in_oid)
{
	PGconn *conn = pg_get_pgconn(self);
	Oid oid = NUM2UINT(in_oid);
	int ret;

	BLOCKING_BEGIN(conn)
		ret = lo_unlink(conn,oid);
	BLOCKING_END(conn)

	if(ret < 0)
		pg_raise_conn_error( rb_ePGerror, self, "lo_unlink failed");

	return Qnil;
}

# File 'ext/pg_connection.c', line 3861

static VALUE
pgconn_lowrite(VALUE self, VALUE in_lo_desc, VALUE buffer)
{
	int n;
	PGconn *conn = pg_get_pgconn(self);
	int fd = NUM2INT(in_lo_desc);

	Check_Type(buffer, T_STRING);

	if( RSTRING_LEN(buffer) < 0) {
		pg_raise_conn_error( rb_ePGerror, self, "write buffer zero string");
	}
	BLOCKING_BEGIN(conn)
		n = lo_write(conn, fd, StringValuePtr(buffer),
				RSTRING_LEN(buffer));
	BLOCKING_END(conn)

	if(n < 0) {
		pg_raise_conn_error( rb_ePGerror, self, "lo_write failed: %s", PQerrorMessage(conn));
	}

	return INT2FIX(n);
}

# File 'ext/pg_connection.c', line 1593

static VALUE
pgconn_make_empty_pgresult(VALUE self, VALUE status)
{
	PGresult *result;
	VALUE rb_pgresult;
	PGconn *conn = pg_get_pgconn(self);
	result = PQmakeEmptyPGresult(conn, NUM2INT(status));
	rb_pgresult = pg_new_result(result, self);
	pg_result_check(rb_pgresult);
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 2236

static VALUE
pgconn_notifies(VALUE self)
{
	t_pg_connection *this = pg_get_connection_safe( self );
	PGnotify *notification;
	VALUE hash;
	VALUE sym_relname, sym_be_pid, sym_extra;
	VALUE relname, be_pid, extra;

	sym_relname = ID2SYM(rb_intern("relname"));
	sym_be_pid = ID2SYM(rb_intern("be_pid"));
	sym_extra = ID2SYM(rb_intern("extra"));

	notification = gvl_PQnotifies(this->pgconn);
	if (notification == NULL) {
		return Qnil;
	}

	hash = rb_hash_new();
	relname = rb_str_new2(notification->relname);
	be_pid = INT2NUM(notification->be_pid);
	extra = rb_str_new2(notification->extra);
	PG_ENCODING_SET_NOCHECK( relname, this->enc_idx );
	PG_ENCODING_SET_NOCHECK( extra, this->enc_idx );

	rb_hash_aset(hash, sym_relname, relname);
	rb_hash_aset(hash, sym_be_pid, be_pid);
	rb_hash_aset(hash, sym_extra, extra);

	PQfreemem(notification);
	return hash;
}

# File 'ext/pg_connection.c', line 2521

static VALUE
pgconn_wait_for_notify(int argc, VALUE *argv, VALUE self)
{
	t_pg_connection *this = pg_get_connection_safe( self );
	PGnotify *pnotification;
	struct timeval timeout;
	struct timeval *ptimeout = NULL;
	VALUE timeout_in = Qnil, relname = Qnil, be_pid = Qnil, extra = Qnil;
	double timeout_sec;

	rb_scan_args( argc, argv, "01", &timeout_in );

	if ( RTEST(timeout_in) ) {
		timeout_sec = NUM2DBL( timeout_in );
		timeout.tv_sec = (time_t)timeout_sec;
		timeout.tv_usec = (suseconds_t)( (timeout_sec - (long)timeout_sec) * 1e6 );
		ptimeout = &timeout;
	}

	pnotification = (PGnotify*) wait_socket_readable( self, ptimeout, notify_readable);

	/* Return nil if the select timed out */
	if ( !pnotification ) return Qnil;

	relname = rb_str_new2( pnotification->relname );
	PG_ENCODING_SET_NOCHECK( relname, this->enc_idx );
	be_pid = INT2NUM( pnotification->be_pid );
	if ( *pnotification->extra ) {
		extra = rb_str_new2( pnotification->extra );
		PG_ENCODING_SET_NOCHECK( extra, this->enc_idx );
	}
	PQfreemem( pnotification );

	if ( rb_block_given_p() )
		rb_yield_values( 3, relname, be_pid, extra );

	return relname;
}

# File 'ext/pg_connection.c', line 747

static VALUE
pgconn_options(VALUE self)
{
	char *options = PQoptions(pg_get_pgconn(self));
	if (!options) return Qnil;
	return rb_str_new2(options);
}

# File 'ext/pg_connection.c', line 835

static VALUE
pgconn_parameter_status(VALUE self, VALUE param_name)
{
	const char *ret = PQparameterStatus(pg_get_pgconn(self), StringValueCStr(param_name));
	if(ret == NULL)
		return Qnil;
	else
		return rb_str_new2(ret);
}

# File 'ext/pg_connection.c', line 660

static VALUE
pgconn_pass(VALUE self)
{
	char *user = PQpass(pg_get_pgconn(self));
	if (!user) return Qnil;
	return rb_str_new2(user);
}

# File 'ext/pg_connection.c', line 3603

static VALUE
pgconn_pipeline_status(VALUE self)
{
	int res = PQpipelineStatus(pg_get_pgconn(self));
	return INT2FIX(res);
}

# File 'ext/pg_connection.c', line 3667

static VALUE
pgconn_pipeline_sync(VALUE self)
{
	PGconn *conn = pg_get_pgconn(self);
	int res = PQpipelineSync(conn);
	if( res != 1 )
		pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));

	return Qnil;
}

# File 'ext/pg_connection.c', line 719

static VALUE
pgconn_port(VALUE self)
{
	char* port = PQport(pg_get_pgconn(self));
	if (!port || port[0] == '\0')
		return INT2NUM(DEF_PGPORT);
	else
		return INT2NUM(atoi(port));
}

# File 'ext/pg_connection.c', line 3391

static VALUE
pgconn_async_prepare(int argc, VALUE *argv, VALUE self)
{
	VALUE rb_pgresult = Qnil;

	pgconn_discard_results( self );
	pgconn_send_prepare( argc, argv, self );
	rb_pgresult = pgconn_async_get_last_result( self );

	if ( rb_block_given_p() ) {
		return rb_ensure( rb_yield, rb_pgresult, pg_result_clear, rb_pgresult );
	}
	return rb_pgresult;
}

# File 'ext/pg_connection.c', line 853

static VALUE
pgconn_protocol_version(VALUE self)
{
	return INT2NUM(PQprotocolVersion(pg_get_pgconn(self)));
}

# File 'lib/pg/connection.rb', line 495

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

# File 'lib/pg/connection.rb', line 525

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

# File 'lib/pg/connection.rb', line 567

def reset
	iopts = conninfo_hash.compact
	if iopts[:host] && !iopts[:host].empty? && PG.library_version >= 100000
		iopts = self.class.send(:resolve_hosts, iopts)
	end
	conninfo = self.class.parse_connect_args( iopts );
	reset_start2(conninfo)
	async_connect_or_reset(:reset_poll)
	self
end

# File 'ext/pg_connection.c', line 614

static VALUE
pgconn_reset_poll(VALUE self)
{
	PostgresPollingStatusType status;

	pgconn_close_socket_io(self);
	status = gvl_PQresetPoll(pg_get_pgconn(self));

	return INT2FIX((int)status);
}

# File 'ext/pg_connection.c', line 597

static VALUE
pgconn_reset_start(VALUE self)
{
	pgconn_close_socket_io( self );
	if(gvl_PQresetStart(pg_get_pgconn(self)) == 0)
		pg_raise_conn_error( rb_eUnableToSend, self, "reset has failed");
	return Qnil;
}