class Connection

Represents a Redis protocol connection handling low-level communication.

Nested Classes and Modules

class << self

Definitions

def initialize(stream)

Initialize a new connection with the provided stream.

Signature
parameter　stream　IO

The underlying stream for communication.

Implementation

 
def initialize(stream)
	@stream = stream
	
	# Number of requests sent:
	@count = 0
end

attr :count

def close

Close the underlying stream connection.

Implementation

 
def close
	@stream.close
end

def flush

Flush any buffered data to the stream.

Implementation

 
def flush
	@stream.flush
end

def closed?

Check if the connection is closed.

Signature
returns　Boolean

True if the connection is closed.

Implementation

 
def closed?
	@stream.closed?
end

def write_request(arguments)

The redis server doesn't want actual objects (e.g. integers) but only bulk strings. So, we inline it for performance.

Implementation

 
def write_request(arguments)
	write_lines("*#{arguments.size}")
	
	@count += 1
	
	arguments.each do |argument|
		string = argument.to_s
		
		write_lines("$#{string.bytesize}", string)
	end
end

def write_object(object)

Write a Redis object to the stream.

Signature
parameter　object　Object

The object to write (String, Array, Integer, or object with to_redis method).

Implementation

 
def write_object(object)
	case object
	when String
		write_lines("$#{object.bytesize}", object)
	when Array
		write_array(object)
	when Integer
		write_lines(":#{object}")
	when nil
		write_lines("$-1")
	else
		write_object(object.to_redis)
	end
end

def write_array(array)

Write a Redis array to the stream.

Signature
parameter　array　Array

The array to write.

Implementation

 
def write_array(array)
	write_lines("*#{array.size}")
	
	array.each do |element|
		write_object(element)
	end
end

def read_data(length)

Read data of specified length from the stream.

Signature
parameter　length　Integer

The number of bytes to read.

returns　String

The data read from the stream.

Implementation

 
def read_data(length)
	buffer = @stream.read(length) or @stream.eof!
	
	# Eat trailing whitespace because length does not include the CRLF:
	@stream.read(2) or @stream.eof!
	
	return buffer
end

def read_object

Read and parse a Redis object from the stream.

Signature
returns　Object

The parsed Redis object (String, Array, Integer, or nil).

raises　ServerError

If the server returns an error response.

raises　EOFError

If the stream reaches end of file.

Implementation

 
def read_object
	# Whether the current response was fully consumed:
	complete = false
	object = nil
	
	# The line that we are processing:
	line = read_line or raise EOFError
	token = line.slice!(0, 1)
	
	case token
	when "$"
		length = line.to_i
		
		if length == -1
			# No data.
		else
			object = read_data(length)
		end
		
		complete = true
	when "*"
		count = line.to_i
		
		if count == -1
			# Null array (https://redis.io/topics/protocol#resp-arrays).
		else
			object = Array.new(count){read_object}
		end
		
		complete = true
	when ":"
		object = line.to_i
		complete = true
	when "-"
		complete = true
		raise ServerError.new(line)
	when "+"
		object = line
		complete = true
	else
		@stream.flush
		
		raise UnknownTokenError, token.inspect
	end
	
	return object
ensure
	close unless complete
end

def write_lines(*arguments)

In the case of Redis, we do not want to perform a flush in every line, because each Redis command contains several lines. Flushing once per command is more efficient because it avoids unnecessary writes to the socket.

Implementation

 
def write_lines(*arguments)
	if arguments.empty?
		@stream.write(CRLF)
	else
		arguments.each do |arg|
			@stream.write(arg)
			@stream.write(CRLF)
		end
	end
end