Class
PhusionPassenger::MessageChannel

In: | lib/phusion_passenger/message_channel.rb
Parent: | Object

This class provides convenience methods for:

- sending and receiving raw data over an IO channel.
- sending and receiving messages over an IO channel.
- file descriptor (IO object) passing over a Unix socket.

All of these methods use exceptions for error reporting.

There are two kinds of messages:

Array messages
  These are just a list of strings, and the message itself has a specific length. The contained strings may not contain NUL characters (`’\0‘`). Note that an array message must have at least one element.
Scalar messages
  These are byte strings which may contain arbitrary binary data. Scalar messages also have a specific length.

The protocol is designed to be low overhead, easy to implement and easy to parse.

MessageChannel is to be wrapped around an IO object. For example:

     a, b = IO.pipe
     channel1 = MessageChannel.new(a)
     channel2 = MessageChannel.new(b)

     # Send an array message.
     channel2.write("hello", "world !!")
     channel1.read    # => ["hello", "world !!"]

     # Send a scalar message.
     channel2.write_scalar("some long string which can contain arbitrary binary data")
     channel1.read_scalar

The life time of a MessageChannel is independent from that of the wrapped IO object. If a MessageChannel object is destroyed, the underlying IO object is not automatically closed. Call close() if you want to close the underlying IO object.

Note: Be careful with mixing the sending/receiving of array messages, scalar messages and IO objects. If you send a collection of any of these in a specific order, then the receiving side must receive them in the exact some order. So suppose you first send a message, then an IO object, then a scalar, then the receiving side must first receive a message, then an IO object, then a scalar. If the receiving side does things in the wrong order then bad things will happen.

Methods

- close
- closed?
- fileno
- new
- read
- read_hash
- read_scalar
- recv_io
- send_io
- write
- write_scalar

Classes and Modules
Class PhusionPassenger::MessageChannel::InvalidHashError
Attributes

[RW] | io | The wrapped IO object.

Public Class methods
new(io = nil)

Create a new MessageChannel by wrapping the given IO object.

[ show source ]

        # File lib/phusion_passenger/message_channel.rb, line 88
    88:         def initialize(io = nil)
    89:                 @io = io
    90:                 # Make it binary just in case.
    91:                 @io.binmode if @io
    92:         end

Public Instance methods
close()

Close the underlying IO stream. Might raise SystemCallError or IOError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 360
    360:         def close
    361:                 @io.close
    362:         end

closed?()

Checks whether the underlying IO stream is closed.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 365
    365:         def closed?
    366:                 return @io.closed?
    367:         end

fileno()

Return the file descriptor of the underlying IO object.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 354
    354:         def fileno
    355:                 return @io.fileno
    356:         end

read()

Read an array message from the underlying file descriptor. Returns the array message as an array, or nil when end-of-stream has been reached.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 100
    100:         def read
    101:                 buffer = new_buffer
    102:                 if !@io.read(HEADER_SIZE, buffer)
    103:                         return nil
    104:                 end
    105:                 while buffer.size < HEADER_SIZE
    106:                         tmp = @io.read(HEADER_SIZE - buffer.size)
    107:                         if tmp.empty?
    108:                                 return nil
    109:                         else
    110:                                 buffer << tmp
    111:                         end
    112:                 end
    113:
    114:                 chunk_size = buffer.unpack(UINT16_PACK_FORMAT)[0]
    115:                 if !@io.read(chunk_size, buffer)
    116:                         return nil
    117:                 end
    118:                 while buffer.size < chunk_size
    119:                         tmp = @io.read(chunk_size - buffer.size)
    120:                         if tmp.empty?
    121:                                 return nil
    122:                         else
    123:                                 buffer << tmp
    124:                         end
    125:                 end
    126:
    127:                 message = []
    128:                 offset = 0
    129:                 delimiter_pos = buffer.index(DELIMITER, offset)
    130:                 while !delimiter_pos.nil?
    131:                         if delimiter_pos == 0
    132:                                 message << ""
    133:                         else
    134:                                 message << buffer[offset .. delimiter_pos - 1]
    135:                         end
    136:                         offset = delimiter_pos + 1
    137:                         delimiter_pos = buffer.index(DELIMITER, offset)
    138:                 end
    139:                 return message
    140:         rescue Errno::ECONNRESET
    141:                 return nil
    142:         end

read_hash()

Read an array message from the underlying file descriptor and return the result as a hash instead of an array. This assumes that the array message has an even number of elements. Returns nil when end-of-stream has been reached.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 151
    151:         def read_hash
    152:                 buffer = new_buffer
    153:                 if !@io.read(HEADER_SIZE, buffer)
    154:                         return nil
    155:                 end
    156:                 while buffer.size < HEADER_SIZE
    157:                         tmp = @io.read(HEADER_SIZE - buffer.size)
    158:                         if tmp.empty?
    159:                                 return nil
    160:                         else
    161:                                 buffer << tmp
    162:                         end
    163:                 end
    164:
    165:                 chunk_size = buffer.unpack(UINT16_PACK_FORMAT)[0]
    166:                 if !@io.read(chunk_size, buffer)
    167:                         return nil
    168:                 end
    169:                 while buffer.size < chunk_size
    170:                         tmp = @io.read(chunk_size - buffer.size)
    171:                         if tmp.empty?
    172:                                 return nil
    173:                         else
    174:                                 buffer << tmp
    175:                         end
    176:                 end
    177:
    178:                 result = {}
    179:                 offset = 0
    180:                 delimiter_pos = buffer.index(DELIMITER, offset)
    181:                 while !delimiter_pos.nil?
    182:                         if delimiter_pos == 0
    183:                                 name = ""
    184:                         else
    185:                                 name = buffer[offset .. delimiter_pos - 1]
    186:                         end
    187:
    188:                         offset = delimiter_pos + 1
    189:                         delimiter_pos = buffer.index(DELIMITER, offset)
    190:                         if delimiter_pos.nil?
    191:                                 raise InvalidHashError
    192:                         elsif delimiter_pos == 0
    193:                                 value = ""
    194:                         else
    195:                                 value = buffer[offset .. delimiter_pos - 1]
    196:                         end
    197:
    198:                         result[name] = value
    199:                         offset = delimiter_pos + 1
    200:                         delimiter_pos = buffer.index(DELIMITER, offset)
    201:                 end
    202:                 return result
    203:         rescue Errno::ECONNRESET
    204:                 return nil
    205:         end

read_scalar(buffer = new_buffer, max_size = nil)

Read a scalar message from the underlying IO object. Returns the read message, or nil on end-of-stream.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

The `buffer` argument specifies a buffer in which #read_scalar stores the read data. It is good practice to reuse existing buffers in order to minimize stress on the garbage collector.

The `max_size` argument allows one to specify the maximum allowed size for the scalar message. If the received scalar message‘s size is larger than `max_size`, then a SecurityError will be raised.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 220
    220:         def read_scalar(buffer = new_buffer, max_size = nil)
    221:                 if !@io.read(4, buffer)
    222:                         return nil
    223:                 end
    224:                 while buffer.size < 4
    225:                         tmp = @io.read(4 - buffer.size)
    226:                         if tmp.empty?
    227:                                 return nil
    228:                         else
    229:                                 buffer << tmp
    230:                         end
    231:                 end
    232:
    233:                 size = buffer.unpack(UINT32_PACK_FORMAT)[0]
    234:                 if size == 0
    235:                         buffer.replace('')
    236:                         return buffer
    237:                 else
    238:                         if !max_size.nil? && size > max_size
    239:                                 raise SecurityError, "Scalar message size (#{size}) " <<
    240:                                         "exceeds maximum allowed size (#{max_size})."
    241:                         end
    242:                         if !@io.read(size, buffer)
    243:                                 return nil
    244:                         end
    245:                         if buffer.size < size
    246:                                 tmp = ''
    247:                                 while buffer.size < size
    248:                                         if !@io.read(size - buffer.size, tmp)
    249:                                                 return nil
    250:                                         else
    251:                                                 buffer << tmp
    252:                                         end
    253:                                 end
    254:                         end
    255:                         return buffer
    256:                 end
    257:         rescue Errno::ECONNRESET
    258:                 return nil
    259:         end

recv_io(klass = IO, negotiate = true)

Receive an IO object (a file descriptor) from the channel. The other side must have sent an IO object by calling send_io(). Note that this only works on Unix sockets.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 297
    297:         def recv_io(klass = IO, negotiate = true)
    298:                 write("pass IO") if negotiate
    299:                 io = @io.recv_io(klass)
    300:                 write("got IO") if negotiate
    301:                 return io
    302:         end

send_io(io)

Send an IO object (a file descriptor) over the channel. The other side must receive the IO object by calling recv_io(). Note that this only works on Unix sockets.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 310
    310:         def send_io(io)
    311:                 # We read a message before actually calling #send_io
    312:                 # in order to prevent the other side from accidentally
    313:                 # read()ing past the normal data and reading our file
    314:                 # descriptor too.
    315:                 #
    316:                 # For example suppose that side A looks like this:
    317:                 #
    318:                 #   read(fd, buf, 1024)
    319:                 #   read_io(fd)
    320:                 #
    321:                 # and side B:
    322:                 #
    323:                 #   write(fd, buf, 100)
    324:                 #   send_io(fd_to_pass)
    325:                 #
    326:                 # If B completes both write() and send_io(), then A's read() call
    327:                 # reads past the 100 bytes that B sent. On some platforms, like
    328:                 # Linux, this will cause read_io() to fail. And it just so happens
    329:                 # that Ruby's IO#read method slurps more than just the given amount
    330:                 # of bytes.
    331:                 result = read
    332:                 if !result
    333:                         raise EOFError, "End of stream"
    334:                 elsif result != ["pass IO"]
    335:                         raise IOError, "IO passing pre-negotiation header expected"
    336:                 else
    337:                         @io.send_io(io)
    338:                         # Once you've sent the IO you expect to be able to close it on the
    339:                         # sender's side, even if the other side hasn't read the IO yet.
    340:                         # Not so: on some operating systems (I'm looking at you OS X) this
    341:                         # can cause the receiving side to receive a bad file descriptor.
    342:                         # The post negotiation protocol ensures that we block until the
    343:                         # other side has really received the IO.
    344:                         result = read
    345:                         if !result
    346:                                 raise EOFError, "End of stream"
    347:                         elsif result != ["got IO"]
    348:                                 raise IOError, "IO passing post-negotiation header expected"
    349:                         end
    350:                 end
    351:         end

write(name, *args)

Send an array message, which consists of the given elements, over the underlying file descriptor. name is the first element in the message, and args are the other elements. These arguments will internally be converted to strings by calling to_s().

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 268
    268:         def write(name, *args)
    269:                 check_argument(name)
    270:                 args.each do |arg|
    271:                         check_argument(arg)
    272:                 end
    273:
    274:                 message = "#{name}#{DELIMITER}"
    275:                 args.each do |arg|
    276:                         message << arg.to_s << DELIMITER
    277:                 end
    278:                 @io.write([message.size].pack('n') << message)
    279:                 @io.flush
    280:         end

write_scalar(data)

Send a scalar message over the underlying IO object.

Might raise SystemCallError, IOError or SocketError when something goes wrong.

[ show source ]

         # File lib/phusion_passenger/message_channel.rb, line 286
    286:         def write_scalar(data)
    287:                 @io.write([data.size].pack('N') << data)
    288:                 @io.flush
    289:         end