class Channel

parameter　io　IO

The connected stream.

def initialize(io)
	@io = io
	@write_mutex = Mutex.new
	@read_closed = false
	@write_closed = false
end

parameter　type　Symbol

The message type.

parameter　payload　Object

The message payload.

def write(type, payload = nil)
	data = Marshal.dump([type, payload])
	
	if data.bytesize > MAXIMUM_FRAME_SIZE
		raise ArgumentError, "Frame is too large: #{data.bytesize} bytes!"
	end
	
	frame = [data.bytesize].pack(HEADER_FORMAT) << data
	
	@write_mutex.synchronize do
		raise ClosedError if @write_closed
		
		write_all(frame)
	end
	
	return nil
rescue IOError, SystemCallError => error
	raise ClosedError, error.message
end

returns　Array(Symbol, Object) | Nil

The next message, or nil after a clean close.

def read
	return nil if @read_closed
	
	header = read_exactly(HEADER_SIZE, eof: true)
	unless header
		@read_closed = true
		return nil
	end
	
	length = header.unpack1(HEADER_FORMAT)
	if length > MAXIMUM_FRAME_SIZE
		raise ClosedError, "Invalid frame size: #{length} bytes!"
	end
	
	return Marshal.load(read_exactly(length))
rescue EOFError, IOError, SystemCallError => error
	raise ClosedError, error.message
end

def close_read
	return if @read_closed
	
	@read_closed = true
	@io.shutdown(::Socket::SHUT_RD)
rescue IOError, SystemCallError
	# The channel was already closed concurrently:
end

def close_write
	@write_mutex.synchronize do
		return if @write_closed
		
		@write_closed = true
		@io.shutdown(::Socket::SHUT_WR)
	end
rescue IOError, SystemCallError
	# The channel was already closed concurrently:
end

def close
	return if @io.closed?
	
	@read_closed = true
	@write_closed = true
	@io.close
rescue IOError
	# The channel was already closed concurrently:
end

returns　Boolean

Whether the channel has been closed.

def closed?
	(@read_closed && @write_closed) || @io.closed?
end

def write_all(buffer)
	offset = 0
	
	while offset < buffer.bytesize
		offset += @io.write(buffer.byteslice(offset, buffer.bytesize - offset))
	end
end

def read_exactly(length, eof: false)
	buffer = String.new(capacity: length, encoding: Encoding::BINARY)
	
	while buffer.bytesize < length
		begin
			buffer << @io.readpartial(length - buffer.bytesize)
		rescue EOFError
			if eof && buffer.empty?
				return nil
			end
			
			raise
		end
	end
	
	return buffer
end