Class: Arf::Wire::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/arf/wire/server.rb,
lib/arf/wire/server/peer.rb

Defined Under Namespace

Classes: Peer

Instance Method Summary collapse

Constructor Details

#initialize(handler) ⇒ Server

Returns a new instance of Server.



8
9
10
11
12
13
14
 
# File 'lib/arf/wire/server.rb', line 8

def initialize(handler)
  @peers = {}
  @peer_id = 0
  @peers_monitor = Monitor.new
  @handler = handler
  @logger = Arf.logger.with_fields(subsystem: "Server")
end

Instance Method Details

#cancel_stream(str) ⇒ Object 



43
 
# File 'lib/arf/wire/server.rb', line 43

def cancel_stream(str) = @handler&.cancel_stream(str)

#handle_stream(str) ⇒ Object 



38
39
40
41
 
# File 'lib/arf/wire/server.rb', line 38

def handle_stream(str)
  @logger.debug("Handle Stream received")
  @handler&.handle_stream(str)
end

#peer_by_id(id) ⇒ Object 



32
33
34
35
36
 
# File 'lib/arf/wire/server.rb', line 32

def peer_by_id(id)
  @peers_monitor.synchronize do
    @peers[id]
  end
end

#register_peer(peer) ⇒ Object 



16
17
18
19
20
21
22
23
24
 
# File 'lib/arf/wire/server.rb', line 16

def register_peer(peer)
  @logger.debug("Registered peer", peer: peer.class.name)
  @peers_monitor.synchronize do
    id = @peer_id
    @peer_id += 1
    @peers[id] = peer
    id
  end
end

#run ⇒ Object 



55
56
57
58
59
60
 
# File 'lib/arf/wire/server.rb', line 55

def run
  config = Arf.config
  @tcp_server = TCPServer.new(config.bind_address, config.bind_port)
  @tcp_server.setsockopt(Socket::SOL_SOCKET, Socket::SO_REUSEADDR, true)
  Arf::Reactor.attach_server(@tcp_server, Peer, self)
end

#shutdown ⇒ Object 



45
46
47
48
49
50
51
52
53
 
# File 'lib/arf/wire/server.rb', line 45

def shutdown
  return unless @tcp_server

  @tcp_server.close
  Arf::Reactor.detach(@tcp_server)
  @peers_monitor.synchronize do
    @peers.each_value { _1.go_away! ERROR_CODE_NO_ERROR, terminate: true }
  end
end

#unregister_peer(id) ⇒ Object 



26
27
28
29
30
 
# File 'lib/arf/wire/server.rb', line 26

def unregister_peer(id)
  @peers_monitor.synchronize do
    @peers.delete(id)
  end
end