Module: Siding::Protocol

Defined in:
lib/siding/protocol.rb

Defined Under Namespace

Classes: Message, MessageTooLarge, ProtocolError, TruncatedMessage, VersionMismatch

Constant Summary collapse

VERSION = 
1
LENGTH_BYTES = 
4
LENGTH_FORMAT = 
"N"
MAX_MESSAGE_BYTES = 
1024 * 1024
HELLO =

Client -> server.

"Hello"
RUN = 
"Run"
SIGNAL = 
"Signal"
STATUS = 
"Status"
STOP = 
"Stop"
WELCOME =

Server -> client.

"Welcome"
VERSION_MISMATCH = 
"VersionMismatch"
BOOTING = 
"Booting"
BOOT_FAILED = 
"BootFailed"
STARTED = 
"Started"
FINISHED = 
"Finished"
STATUS_REPORT = 
"StatusReport"
STREAM_ORDER = 
%i[stdin stdout stderr].freeze

Class Method Summary collapse

Class Method Details

.client_handshake(io) ⇒ Object

Raises:



125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/siding/protocol.rb', line 125

def client_handshake(io)
  write_message(io, HELLO, protocol_version: VERSION)
  reply = read_message(io)
  raise TruncatedMessage, "server closed during handshake" if reply.nil?

  case reply.type
  when WELCOME
    reply["protocol_version"]
  when VERSION_MISMATCH
    raise VersionMismatch.new(server_version: reply["protocol_version"])
  else
    raise ProtocolError, "unexpected #{reply.type.inspect} in response to #{HELLO}"
  end
end

.read_exactly(io, count, allow_eof: false) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/siding/protocol.rb', line 85

def read_exactly(io, count, allow_eof: false)
  return +"" if count.zero?

  buffer = +""
  while buffer.bytesize < count
    chunk = read_some(io, count - buffer.bytesize)
    if chunk.nil? || chunk.empty?
      return nil if buffer.empty? && allow_eof

      raise TruncatedMessage, "peer sent #{buffer.bytesize} of #{count} bytes before closing"
    end
    buffer << chunk
  end
  buffer
end

.read_message(io) ⇒ Object

Raises:



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/siding/protocol.rb', line 65

def read_message(io)
  header = read_exactly(io, LENGTH_BYTES, allow_eof: true)
  return nil if header.nil?

  length = header.unpack1(LENGTH_FORMAT)
  raise MessageTooLarge, "peer announced #{length} bytes, over the #{MAX_MESSAGE_BYTES} limit" if
    length > MAX_MESSAGE_BYTES

  body = read_exactly(io, length)
  parsed = begin
    JSON.parse(body)
  rescue JSON::ParserError => e
    raise ProtocolError, "malformed message body: #{e.message}"
  end

  raise ProtocolError, "message has no type" unless parsed.is_a?(Hash) && parsed["type"]

  Message.new(type: parsed.delete("type"), payload: parsed)
end

.read_some(io, count) ⇒ Object 



101
102
103
 
# File 'lib/siding/protocol.rb', line 101

def read_some(io, count)
  io.respond_to?(:recv) ? io.recv(count) : io.read(count)
end

.receive_streams(io) ⇒ Object 



116
117
118
119
120
121
122
123
 
# File 'lib/siding/protocol.rb', line 116

def receive_streams(io)
  STREAM_ORDER.each_with_object({}) { |name, streams| streams[name] = io.recv_io }
rescue EOFError, SocketError, SystemCallError => e
  # A peer that dies mid-pass surfaces as `SocketError` ("file descriptor was not passed")
  # rather than as EOF, because the control message arrives empty rather than not at all.
  # Both mean the same thing here: the descriptors we were promised are not coming.
  raise TruncatedMessage, "peer closed while passing descriptors: #{e.message}"
end

.send_streams(io, stdin: $stdin, stdout: $stdout, stderr: $stderr) ⇒ Object 



111
112
113
114
 
# File 'lib/siding/protocol.rb', line 111

def send_streams(io, stdin: $stdin, stdout: $stdout, stderr: $stderr)
  [stdin, stdout, stderr].each { |stream| io.send_io(stream) }
  STREAM_ORDER
end

.server_handshake(io) ⇒ Object

Raises:



140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/siding/protocol.rb', line 140

def server_handshake(io)
  hello = read_message(io)
  raise TruncatedMessage, "client closed during handshake" if hello.nil?
  raise ProtocolError, "expected #{HELLO}, got #{hello.type.inspect}" unless hello.type == HELLO

  if hello["protocol_version"] == VERSION
    write_message(io, WELCOME, protocol_version: VERSION)
    true
  else
    write_message(io, VERSION_MISMATCH, protocol_version: VERSION)
    false
  end
end

.stringify(payload) ⇒ Object 



105
106
107
 
# File 'lib/siding/protocol.rb', line 105

def stringify(payload)
  payload.each_with_object({}) { |(k, v), h| h[k.to_s] = v }
end

.write_message(io, type, payload = {}) ⇒ Object

Raises:



53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/siding/protocol.rb', line 53

def write_message(io, type, payload = {})
  body = JSON.generate({ "type" => type }.merge(stringify(payload)))
  raise MessageTooLarge, "message of #{body.bytesize} bytes exceeds #{MAX_MESSAGE_BYTES}" if body.bytesize > MAX_MESSAGE_BYTES

  # Length and body in a single write. Two writes would give a reader a window in which a
  # crash leaves a valid-looking prefix with nothing behind it -- exactly the state the
  # framing is here to make unrepresentable.
  io.write([body.bytesize].pack(LENGTH_FORMAT) + body)
  io.flush if io.respond_to?(:flush)
  body.bytesize
end