Module: RSMP::SiteConnections

Included in:
Site
Defined in:
lib/rsmp/node/site/connections.rb

Overview

Connection handling for site instances.

Instance Method Summary collapse

Instance Method Details

#accept_supervisor_connection(socket) ⇒ Object 



38
39
40
41
42
43
44
45
46
 
# File 'lib/rsmp/node/site/connections.rb', line 38

def accept_supervisor_connection(socket)
  remote_port = socket.remote_address.ip_port
  remote_ip = socket.remote_address.ip_address
  proxy = SupervisorProxy.new(accepted_supervisor_options(socket, remote_ip, remote_port))
  @proxies << proxy
  @proxies_condition.signal
  proxy.start
  proxy.wait
end

#accept_supervisor_connections ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/rsmp/node/site/connections.rb', line 26

def accept_supervisor_connections
  @endpoint.accept do |socket|
    accept_supervisor_connection socket
  rescue StandardError => e
    distribute_error e, level: :internal
  end
rescue Async::Stop
  # Expected during shutdown - no action needed
rescue StandardError => e
  distribute_error e, level: :internal
end

#build_proxies ⇒ Object 



4
5
6
7
8
9
10
 
# File 'lib/rsmp/node/site/connections.rb', line 4

def build_proxies
  return if server_role?

  @site_settings['supervisors'].each do |supervisor_settings|
    @proxies << SupervisorProxy.new(supervisor_proxy_options(supervisor_settings))
  end
end

#connect_to_supervisor(_task, supervisor_settings) ⇒ Object 



48
49
50
51
52
53
 
# File 'lib/rsmp/node/site/connections.rb', line 48

def connect_to_supervisor(_task, supervisor_settings)
  proxy = build_proxy(supervisor_proxy_options(supervisor_settings))
  @proxies << proxy
  proxy.start
  @proxies_condition.signal
end

#listen_for_supervisors ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/rsmp/node/site/connections.rb', line 12

def listen_for_supervisors
  ip = @site_settings['ip'] || '0.0.0.0'
  port = @site_settings['port']
  log "Starting #{site_type_name} listener on #{ip}:#{port}", level: :info, timestamp: @clock.now
  @endpoint = IO::Endpoint.tcp(ip, port)
  @accept_task = Async::Task.current.async do |task|
    task.annotate 'site accept loop'
    accept_supervisor_connections
  end

  @ready_condition.signal
  @accept_task.wait
end