Async::RedisSourceAsyncRedisEndpoint

class Endpoint

Represents a way to connect to a remote Redis server.

Definitions

def self.local(**options)

Create a local Redis endpoint.

Signature

parameter options Hash

Additional options for the endpoint.

returns Endpoint

A local Redis endpoint.

Implementation

def self.local(**options)
	self.new(LOCALHOST, **options)
end

def self.remote(host, port = 6379, **options)

Create a remote Redis endpoint.

Signature

parameter host String

The hostname to connect to.

parameter port Integer

The port to connect to.

parameter options Hash

Additional options for the endpoint.

returns Endpoint

A remote Redis endpoint.

Implementation

def self.remote(host, port = 6379, **options)
	# URI::Generic.build automatically handles IPv6 addresses correctly:
	self.new(URI::Generic.build(scheme: "redis", host: host, port: port), **options)
end

def self.parse(string, endpoint = nil, **options)

Parse a Redis URL string into an endpoint.

Signature

parameter string String

The URL string to parse.

parameter endpoint Endpoint

Optional underlying endpoint.

parameter options Hash

Additional options for the endpoint.

returns Endpoint

The parsed endpoint.

Implementation

def self.parse(string, endpoint = nil, **options)
	url = URI.parse(string)
	
	return self.new(url, endpoint, **options)
end

def self.for(scheme, host, credentials: nil, port: nil, database: nil, **options)

Construct an endpoint with a specified scheme, hostname, optional path, and options.

Signature

parameter scheme String

The scheme to use, e.g. "redis" or "rediss".

parameter hostname String

The hostname to connect to (or bind to).

parameter options Hash

Additional options, passed to #initialize.

Implementation

def self.for(scheme, host, credentials: nil, port: nil, database: nil, **options)
	uri_klass = SCHEMES.fetch(scheme.downcase) do
		raise ArgumentError, "Unsupported scheme: #{scheme.inspect}"
	end
	
	if database
		path = "/#{database}"
	end
	
	self.new(
		uri_klass.build(
			scheme: scheme,
			userinfo: credentials&.join(":"),
			host: host,
			port: port,
			path: path,
		),
		**options
	)
end

def self.[](object)

Coerce the given object into an endpoint.

Signature

parameter url String | Endpoint

The URL or endpoint to convert.

Implementation

def self.[](object)
	if object.is_a?(self)
		return object
	else
		self.parse(object.to_s)
	end
end

def initialize(url, endpoint = nil, **options)

Create a new endpoint.

Signature

parameter url URI

The URL to connect to.

parameter endpoint Endpoint

The underlying endpoint to use.

option scheme String

The scheme to use, e.g. "redis" or "rediss".

option hostname String

The hostname to connect to (or bind to), overrides the URL hostname (used for SNI).

option port Integer

The port to bind to, overrides the URL port.

option ssl_context OpenSSL::SSL::SSLContext

The SSL context to use for secure connections.

Implementation

def initialize(url, endpoint = nil, **options)
	super(**options)
	
	raise ArgumentError, "URL must be absolute (include scheme, host): #{url}" unless url.absolute?
	
	@url = url
	
	if endpoint
		@endpoint = self.build_endpoint(endpoint)
	else
		@endpoint = nil
	end
end

def to_url

Convert the endpoint to a URL.

Signature

returns URI

The URL representation of the endpoint.

Implementation

def to_url
	url = @url.dup
	
	unless default_port?
		url.port = self.port
	end
	
	return url
end

def to_s

Convert the endpoint to a string representation.

Signature

returns String

A string representation of the endpoint.

Implementation

def to_s
	"\#<#{self.class} #{self.to_url} #{@options}>"
end

def inspect

Convert the endpoint to an inspectable string.

Signature

returns String

An inspectable string representation of the endpoint.

Implementation

def inspect
	"\#<#{self.class} #{self.to_url} #{@options.inspect}>"
end

def address

Get the address of the underlying endpoint.

Signature

returns String

The address of the endpoint.

Implementation

def address
	endpoint.address
end

def secure?

Check if the connection is secure (using TLS).

Signature

returns Boolean

True if the connection uses TLS.

Implementation

def secure?
	["rediss"].include?(self.scheme)
end

def protocol

Get the protocol for this endpoint.

Signature

returns Protocol

The protocol instance configured for this endpoint.

Implementation

def protocol
	protocol = @options.fetch(:protocol, Protocol::RESP2)
	
	if credentials = self.credentials
		protocol = Protocol::Authenticated.new(credentials, protocol)
	end
	
	if database = self.database
		protocol = Protocol::Selected.new(database, protocol)
	end
	
	return protocol
end

def default_port

Get the default port for Redis connections.

Signature

returns Integer

The default Redis port (6379).

Implementation

def default_port
	6379
end

def default_port?

Check if the endpoint is using the default port.

Signature

returns Boolean

True if using the default port.

Implementation

def default_port?
	port == default_port
end

def port

Get the port for this endpoint.

Signature

returns Integer

The port number.

Implementation

def port
	@options[:port] || @url.port || default_port
end

def hostname

The hostname is the server we are connecting to:

Implementation

def hostname
	@options[:hostname] || @url.hostname
end

def scheme

Get the scheme for this endpoint.

Signature

returns String

The URL scheme (e.g., "redis" or "rediss").

Implementation

def scheme
	@options[:scheme] || @url.scheme
end

def database

Get the database number for this endpoint.

Signature

returns Integer | Nil

The database number or nil if not specified.

Implementation

def database
	@options[:database] || extract_database(@url.path)
end

def ssl_verify_mode

We don't try to validate peer certificates when talking to localhost because they would always be self-signed.

Implementation

def ssl_verify_mode
	if self.localhost?
		OpenSSL::SSL::VERIFY_NONE
	else
		OpenSSL::SSL::VERIFY_PEER
	end
end