ピュアP2Pでルームが1つしかないチャットのソース

rubyp2p.rb

require 'socket'
require 'digest/md5'
require 'thread'
require 'timeout'
$q = Queue.new
$KCODE = 'u'
init_seed = []
$regi_seed = []
$talk_sum = []
if ARGV[0]
  $my_port = ARGV[0].to_i
else
  $my_port = 18782
end
$my_name = "NoName"
$my_name = ARGV[1] if ARGV[1]
open("initseed_"+$my_port.to_s+".txt","r") {|f|
  while l = f.gets
    init_seed.push l.gsub("\n","").split(":")
  end
}
init_seed.each_index {|i|
  init_seed[i][1] = init_seed[i][1].to_i
}

puts "info>My port is "+$my_port.to_s
print "info> My initseed is:"
p init_seed
$sockets = []




#begin
th_serv = Thread.start do
  so_serv = TCPServer.open($my_port)
  puts "info> Server started...."
  while true
    Thread.start(so_serv.accept) do |s|
      $sockets.push s
      p s.peeraddr
      puts "info>"+s.peeraddr[2]+"("+s.peeraddr[3]+") is accepted"
      while s.gets
        n = $_
        if /regist/ =~ n
          n = n.gsub("\n","").gsub("\r","")
          na = n.split(" ")
          #p na
          $regi_seed.push [s.peeraddr[3],na[1].to_i]
          puts "info>"+s.peeraddr[2]+"("+s.peeraddr[3]+"):"+na[1].to_s+" is registed"
          s.write("reg_ok\n")
        elsif /talk/ =~ n
          n = n.gsub("\n","").gsub("\r","")
          na = n.split(":")
          if !$talk_sum.include?(Digest::MD5.new.update(n))
            puts na[1]+": "+na[2]
            #p Digest::MD5.new.update(n)
            $talk_sum.push Digest::MD5.new.update(n)
          end
          $q.push n
        else
          n = n.gsub("\n","").gsub("\r","")
          #p n
          $q.push n
        end
      end
      puts "info>"+s.peeraddr[2]+"("+s.peeraddr[3]+") is gone"
      $regi_seed.delete_at($regi_seed.index([s.peeraddr[2],s.peeraddr[3]]))
      $sockets.delete_at($regi_seed.index(s))
      s.close
    end
  end
end#accept thread
Signal.trap(:INT){

  exit 0
}
puts "info>Join to P2P Network..."

init_seed.each {|n|
  puts "info>Connecting to "+n[0]+":"+n[1].to_s+" ....(timeout=10sec)"
  begin
    s = timeout(10) {TCPSocket.new(n[0],n[1])}
  rescue TimeoutError
    puts "info>Timeout occurred.Skip."
  rescue
    puts "info>Skip"
  else
    s.write("regist "+$my_port.to_s+"\n")
    s.flush
    puts 'info>Waiting for "reg_ok" ....'
    i = 0
    while s.gets != "reg_ok\n"
      #puts 'info>waiting'
      i += 1
    end
    puts 'info>Loop count: '+i.to_s
    $regi_seed.push n
    puts "info>"+n[0]+":"+n[1].to_s+" is registed"
    Thread.new(s){|s|
      $sockets.push s
      while s.gets
        n = $_
        if /talk/ =~ n
          n = n.gsub("\n","").gsub("\r","")
          na = n.split(":")
          if !$talk_sum.include?(Digest::MD5.new.update(n))
            puts na[1]+": "+na[2]
            #p Digest::MD5.new.update(n)
            $talk_sum.push Digest::MD5.new.update(n)
          end
          $q.push n
        else   
          n = n.gsub("\n","").gsub("\r","")
          #p n
          $q.push n
        end
      end
      puts "info>"+s.peeraddr[2]+"("+s.peeraddr[3]+") is gone"
      $regi_seed.delete_at($regi_seed.index([s.peeraddr[2],s.peeraddr[3]]))
      $sockets.delete_at($regi_seed.index(s))
      s.close
    }
  end
}
puts "info>P2P Network joined"
Thread.new{
  sums = []
  while true

    #while !$q.empty?
    tmp = $q.pop
    if tmp && !sums.include?(Digest::MD5.new.update(tmp))
      #print "info>pop "
      #p tmp
      #p $sockets
      #puts "Send"
      sums.push(Digest::MD5.new.update(tmp))
      $sockets.each {|s|
        begin
          #p s.peeraddr
          s.write(tmp+"\n")
          s.flush
        rescue Errno::EPIPE
          #print "info>error:"
          #p $!
          $sockets.delete_at($sockets.index(s))
          #rescue
        rescue
          puts "info>error "
          p $!
        end
      }
    end
    #end
  end
}
#Thread.new{
l = ""
while true
  #print "say>"
  l = $stdin.gets.chop!
  if l == "/exit"
    exit 0
  end
  if l != ""
    $q.push "talk:"+$my_name+":"+l+"\n"
    #puts $my_name+": "+l
  end
end

th_serv.join
#th_serv.join
#rescue
#  print "info> ERROR! "
#  p $!
#  puts "info> Exit"
#end