Celluloid IO bug reproduction

repro.rb

require "celluloid/io"

class Writer
  include Celluloid::IO
  include Celluloid::Logger

  def initialize(host = "localhost", port = 0)
    @server = TCPServer.new(host, port)
    @host = host
    @port = @server.addr[1]

    async.start
  end

  attr_reader :host, :port

  def start
    info "Listening on #{@host}:#{@port}"
    loop { async.handle_client(@server.accept) }
  end

  def handle_client(socket)
    info "Accepting client #{socket}"
    data = DATA.read
    loop { socket << data }
  rescue Errno::EPIPE
    info "Client died."
  end
end

class Reader
  include Celluloid::IO
  include Celluloid::Logger

  def initialize(host, port)
    @host = host
    @port = port
    async.start
  end

  attr_reader :host, :port

  def start
    info "Connecting to #{host}:#{port}"
    TCPSocket.open(host, port) do |socket|
      buffer = "".force_encoding("BINARY")

      loop do
        # Our current workaround to force celluloid to check the mailbox:
        # sleep(0)

        # keep a small buffer to make it less likely we read all of it,
        # just to make the bug easier to see
        socket.readpartial(10, buffer)

        print "."

        # for now we will use kernel sleep to simulate the
        # actor being blocked on working on something long
        # enough for the writer to fill the socket with data
        #
        # if we used something CPU-intensive this example
        # might not work relibably in MRI
        Kernel.sleep(0.2)
      end
    end
  end
end

puts "Starting actors."

server = Writer.new
puts "Giving writer time to start."
sleep 0.2

reader = Reader.new(server.host, server.port)
puts "Giving reader time to start."
sleep 0.2

puts "All should be up. Requesting termination."
reader.terminate

puts "Finished. Bye bye."

__END__
<events>
  <event name="some XML">
    <attribute>value</attribute>
    <attribute>value</attribute>
    <attribute>value</attribute>
    <attribute>value</attribute>
    <attribute>value</attribute>
  </event>