Ione Ping Pong

pingpong.rb

# encoding: utf-8

$: << File.expand_path('../../lib', __FILE__)

require 'ione'
require 'thread'

MESSAGE = 'hello world!'

def setup_server(acceptor)
  acceptor.on_accept do |connection|
    connection.on_data do |data|
      connection.write(data)
    end
  end
end

def setup_client(connection, name)
  queue = Queue.new
  connection.on_data { |data| queue << data }
  Thread.start do
    begin
      connection.write(MESSAGE)
      buffer = Ione::ByteBuffer.new
      message_size = MESSAGE.bytesize
      counter = 0
      started_at = Time.now
      loop do
        buffer << queue.pop
        if buffer.bytesize >= message_size
          buffer.discard(message_size)
          counter += 1
          if counter % 10_000 == 0
            time_taken = Time.now - started_at
            started_at = Time.now
            $stderr.printf("%s %d/s\n", name, 10_000/time_taken)
          end
          connection.write(MESSAGE)
        end
      end
    rescue => e
      $stderr.puts("Bork: #{e.message}")
    end
  end
end

f = Ione::Io::IoReactor.new.start
f = f.flat_map do |reactor|
  ff = reactor.bind('0.0.0.0', 0)
  ff.on_value do |acceptor|
    setup_server(acceptor)
  end
  ff
end
server = f.value

if server.respond_to?(:host)
  host, port = server.host, server.port
elsif (defined? JRUBY_VERSION) && JRUBY_VERSION.start_with?('1.7')
  addrinfo = server.to_io.local_address
  host = addrinfo.ip_address
  port = addrinfo.ip_port
else
  host, port = server.to_io.local_address.ip_unpack
end

f = Ione::Io::IoReactor.new.start
f = f.flat_map do |reactor|
  Ione::Future.traverse(%w[a b c]) do |name|
    ff = reactor.connect(host, port)
    ff.on_value do |connection|
      setup_client(connection, name)
    end
    ff
  end
end
f.value

sleep