Class: Netz::Client

Inherits:
Object
  • Object
show all
Includes:
ManagementCommands
Defined in:
lib/netz/client.rb

Constant Summary collapse

MANAGEMENT_PORT = 
7370
INITIAL_PEER_PORT = 
7374

Constants included from ManagementCommands

ManagementCommands::PEER_CONNECT, ManagementCommands::START

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Client

Returns a new instance of Client.



12
13
14
15
16
17
18
 
# File 'lib/netz/client.rb', line 12

def initialize
  @port = INITIAL_PEER_PORT
  @peers = SafeArray.new
  Thread.new do
    setup_management_port
  end
end

Instance Method Details

#add_peer(peer_socket) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/netz/client.rb', line 33

def add_peer(peer_socket)
  @port += 1
  Thread.new do
    peer_server = TCPServer.open @port 
    puts "waiting for peer on per-peer port #{@port}"
    peer = peer_server.accept
    puts "accepted peer."
    @peers.push(peer)
  end
  #push port, then ip strings
  peer_socket << [@port].pack("I") 
  peer_socket << [@peers.size].pack("I") 
  @peers.each do |peer_socket|
    # TODO will this always be peeraddr? or sometime addr depending on Server vs Socket?
    to_send_address = peer_socket.peeraddr[3].split(":").last
    peer_socket << [to_send_address.length].pack("I") 
    peer_socket << to_send_address
  end
end

#connect_to_peer(ip, port) ⇒ Object 



103
104
105
106
 
# File 'lib/netz/client.rb', line 103

def connect_to_peer(ip, port)
  puts "connecting to peer #{ip}:#{port}"
  @peers << TCPSocket.new(ip, port)
end

#join(ip) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/netz/client.rb', line 108

def join(ip)
  puts "trying to join #{ip}"
  @joined_peer_ips ||= []
  @joined_peer_ips << ip
  peer_accept = TCPSocket.new ip, MANAGEMENT_PORT
  peer_accept << [PEER_CONNECT].pack("C") 

  peer_ips = []
  join_port = peer_accept.recvfrom(4)[0].unpack("I")[0]
  puts "got join_port #{join_port}"
  num_additional_peers = peer_accept.recvfrom(4)[0].unpack("I")[0]
  puts "got num_additional_peers #{num_additional_peers}"
  num_additional_peers.times do
    puts "getting peer ip from stream"
    ip_length = peer_accept.recvfrom(4)[0].unpack("I")
    peer_ip = peer_accept.recvfrom(ip_length)[0].unpack("a")
    peer_ips << peer_ip unless @joined_peer_ips.include? peer_ip
  end

  connect_to_peer ip, join_port

  peer_accept.close

  peer_ips.each do |pip|
    join pip
  end

end

#setup_management_port ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/netz/client.rb', line 70

def setup_management_port
  puts "creating management port on #{MANAGEMENT_PORT}"
  begin
    server = TCPServer.open MANAGEMENT_PORT
    @management_accept_thread = Thread.new do
      begin
        loop do
          puts "management thread trying to accept..."
          peer = server.accept
          puts "YAY! peer connected!"

          cmd = peer.recvfrom(1)[0].unpack("C")[0].to_i
          case cmd
          when PEER_CONNECT
            puts "connection from peer #{peer.inspect}"
            add_peer peer
          when START
            puts "got start"
            setup_with_siblings
          else
            puts "unknown cmd"
            p cmd
          end
        
        end
      end
    end
  rescue Exception => ex
    puts "client failed to open management port: #{ex}"
  end

end

#setup_with_siblings ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/netz/client.rb', line 53

def setup_with_siblings
  puts "setting up siblings"
  @channel = Queue.new
  broadcaster = Broadcaster.new
  broadcaster.command_channel = @channel
  broadcaster.remote_clients = @peers

  catcher = Catcher.new
  catcher.command_channel = @channel
  catcher.remote_clients = @peers

  catcher.run
  broadcaster.run

  @channel.push Command.new
end

#start_all ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/netz/client.rb', line 20

def start_all
  setup_with_siblings

  @peers.each do |peer_socket|
    to_send_address = peer_socket.peeraddr[3].split(":").last
    puts "STARTING... #{to_send_address}"
    peer_mng_socket = TCPSocket.new to_send_address, MANAGEMENT_PORT

    peer_mng_socket << [START].pack("C") 
    peer_mng_socket.close
  end
end