One server, multiple clients, asynchronous bi-directional communication using TCP Sockets

client2.rb

#!/usr/bin/env ruby
#encoding: UTF-8
require_relative 'server2'

class Laink::Client
	attr_reader :name
  def initialize
    @server = nil
    @name   = "Client #{rand(10000).to_s(36)}"
  end

  def connected?
    @server && [email protected]?
  end

  def disconnect
    @server.close
    @server = nil
  end

  def connect( ip="localhost", port=Laink::Server::DEFAULT_PORT )
    disconnect if connected?
    begin
    	socket = TCPSocket.new( ip, port ) #FIXME: timeout
      @server = Laink::JSONSocket.new( socket ) 
    rescue Errno::ECONNREFUSED => e
      warn "Could not connect to #{ip}:#{port}."
      raise
    end
  end

  def handle_message( message )
    puts "I just heard #{message.inspect}"
  end

  def run
    connect unless connected?
    @reader = Thread.new{ loop{ handle_message @server.read_data } }
    5.times{ |i|
      @server.send_data command:"Hello ##{i}", source:name
      sleep rand(3)
    }
  end
end

Laink::Client.new.run if __FILE__==$0

server2.rb

#!/usr/bin/env ruby
#encoding: UTF-8

require 'socket'
require 'json'
module Laink; end

class Laink::JSONSocket
  attr_reader :socket, :address
  def initialize( socket )
    @socket  = socket
    @address = "%s:%i" % @socket.remote_address.ip_unpack
  end
  def send_data( data )
    json  = data.to_json
    bytes = json.length
    unless @socket.closed?
      @socket.write [bytes].pack('n')
      unless @socket.closed?
        @socket.write json
      end
    end
  end
  def read_data
    begin
      bytes = @socket.read(2)
      raise IOError.new("no byte header") unless bytes && bytes.length==2
      bytes = bytes.unpack('n').first
      json  = @socket.read(bytes)
      raise IOError.new("not enough JSON data") unless json && json.length==bytes
      JSON.parse("[#{json}]",symbolize_names:true)[0]
    rescue JSON::ParserError => e
      warn "Failed to parse JSON response: #{json.inspect}; #{e}"
    rescue IOError => e
      @socket.close
    end
  end
  def closed?
    @socket.closed?
  end  
end

class Laink::Server
  DEFAULT_PORT = 54147
  def initialize( port=DEFAULT_PORT )
    server = TCPServer.new(port)
    puts "Starting #{self.class} on %s:%i" % server.local_address.ip_unpack
    loop do
      Thread.start(server.accept) do |socket|
        client = Laink::JSONSocket.new(socket)
        begin
          reader = Thread.new{ loop{
            break if client.closed?
            handle_message( client, client.read_data )
          } }
          10.times{ |i|
            break if client.closed?
            client.send_data command:"G'day ##{i}", source:"Server"
            sleep rand(3)
          }
        ensure
          puts "Closing from #{client.address}"
          socket.close
        end
      end
    end
  end
  def handle_message( client, message )
    puts "I just heard #{message.inspect}" if message
  end
end

Laink::Server.new if __FILE__==$0

z_runtime_results.txt

## terminal 1 (server)
phrogz$ ruby server2.rb 
Starting Laink::Server on 0.0.0.0:54147
I just heard {:command=>"Hello #0", :source=>"Client 2tc"}
I just heard {:command=>"Hello #1", :source=>"Client 2tc"}
I just heard {:command=>"Hello #2", :source=>"Client 2tc"}
I just heard {:command=>"Hello #3", :source=>"Client 2tc"}
I just heard {:command=>"Hello #0", :source=>"Client d3"}
I just heard {:command=>"Hello #4", :source=>"Client 2tc"}
I just heard {:command=>"Hello #1", :source=>"Client d3"}
I just heard {:command=>"Hello #2", :source=>"Client d3"}
Closing from 127.0.0.1:62365
I just heard {:command=>"Hello #3", :source=>"Client d3"}
I just heard {:command=>"Hello #4", :source=>"Client d3"}
Closing from 127.0.0.1:62367

## terminal 2 (client 1)
phrogz$ ruby client2.rb 
I just heard {:command=>"G'day #0", :source=>"Server"}
I just heard {:command=>"G'day #1", :source=>"Server"}
I just heard {:command=>"G'day #2", :source=>"Server"}
I just heard {:command=>"G'day #3", :source=>"Server"}
I just heard {:command=>"G'day #4", :source=>"Server"}
I just heard {:command=>"G'day #5", :source=>"Server"}
I just heard {:command=>"G'day #6", :source=>"Server"}
I just heard {:command=>"G'day #7", :source=>"Server"}
phrogz$ 

## terminal 3 (client 2)
phrogz$ ruby client2.rb 
I just heard {:command=>"G'day #0", :source=>"Server"}
I just heard {:command=>"G'day #1", :source=>"Server"}
I just heard {:command=>"G'day #2", :source=>"Server"}
I just heard {:command=>"G'day #3", :source=>"Server"}
I just heard {:command=>"G'day #4", :source=>"Server"}
phrogz$