Class: Protocol::HTTP::Executor::Channel

Inherits:
Object
  • Object
show all
Defined in:
lib/protocol/http/executor/channel.rb

Overview

A framed, bidirectional message channel over an IO object.

Constant Summary collapse

HEADER_FORMAT = 
"N"
HEADER_SIZE = 
4
MAXIMUM_FRAME_SIZE = 
256 * 1024 * 1024

Instance Method Summary collapse

Constructor Details

#initialize(io) ⇒ Channel

Initialize a channel over the given IO object.



20
21
22
23
24
25
 
# File 'lib/protocol/http/executor/channel.rb', line 20

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

Instance Method Details

#close ⇒ Object

Close the underlying IO object.



96
97
98
99
100
101
102
103
104
 
# File 'lib/protocol/http/executor/channel.rb', line 96

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

#close_read ⇒ Object

Shut down the reading direction while leaving the writing direction available.



74
75
76
77
78
79
80
81
 
# File 'lib/protocol/http/executor/channel.rb', line 74

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

#close_write ⇒ Object

Shut down the writing direction while leaving the reading direction available.



84
85
86
87
88
89
90
91
92
93
 
# File 'lib/protocol/http/executor/channel.rb', line 84

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

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


107
108
109
 
# File 'lib/protocol/http/executor/channel.rb', line 107

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

#read ⇒ Object

Read the next typed message.



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/protocol/http/executor/channel.rb', line 54

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

#write(type, payload = nil) ⇒ Object

Send a typed message.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/protocol/http/executor/channel.rb', line 31

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