Simple EventMachine TCP messaging framing example

em_tcp_message_framing.rb

require 'bindata'

module ServerConnection
  def receive_data(data)
    @buffer ||= []
    @data_size ||= 0
    @current_size ||= 0

    if @buffer.empty?
      # First four characters of any framed message will always be the size of message
      @data_size = BinData::Uint32be.read(@buffer[0..3]).to_i
      data = data[4..-1]
    end

    take_data_size = @data_size - @current_size
    fragment = data[0..take_data_size]
    @buffer << fragment
    @current_size += fragment.length

    if @current_size == @data_size
      begin
        process_message(@buffer.join) 
      rescue => e
        # make sure you catch your exceptions here or your server will explode!
      ensure
        # if you're catching exceptions, make sure you always clear your buffer in the end
        # otherwise it'll mess with your next framed messaging processing
        @buffer.clear
        @current_size = 0
        @data_size = 0

        # In the case where we received the next framed message,
        # recursively call receive_data to process it
        receive_data(data[(take_data_size + 1)..-1]) if data.length > take_data_size
      end
    end
  end

  def send_message(message)
    # Will always be four characters long
    length = BinData::Uint32be.new(message.length).to_binary_s

    send_data(length+message)
  end

  def process_message(message)
    # do something here to process it
  end
end

module ServerConnection
def receive_data(data)
@buffer ||= []
@data_size ||= 0
@current_size ||= 0

if @buffer.empty?
  # First four characters of any framed message will always be the size of message
  @data_size = BinData::Uint32be.read(@buffer[0..3]).to_i
  data = data[4..-1]
end

take_data_size = @data_size - @current_size
payload = data[0..take_data_size]
@buffer << data
@current_size += payload.length

if @current_size == @data_size
  begin
    process_message(@buffer.join) 
  rescue => e
    # make sure you catch your exceptions here or your server will explode!
  ensure
    # if you're catching exceptions, make sure you always clear your buffer in the end
    # otherwise it'll mess with your next framed messaging processing
    @buffer.clear
    @current_size = 0
    @data_size = 0

    # In the case where we received the next framed message,
    # recursively call receive_data to process it
    receive_data(data[(take_data_size + 1)..-1]) if data.length > take_data_size
  end
end

end

def send_message(message)
# Will always be four characters long
length = BinData::Uint32be.new(message.length).to_binary_s

send_data(length+message)

end

def process_message(message)
# do something here to process it
end
end