130 lines
4.3 KiB
Ruby
130 lines
4.3 KiB
Ruby
require 'em-websocket'
|
|
require 'bugsnag'
|
|
module JamWebsockets
|
|
|
|
class Server
|
|
|
|
def initialize(options={})
|
|
EM::WebSocket.close_timeout = 10 # the default of 60 is pretty intense
|
|
@log = Logging.logger[self]
|
|
@count=0
|
|
@router = Router.new
|
|
@ar_base_logger = ::Logging::Repository.instance[ActiveRecord::Base]
|
|
end
|
|
|
|
def run(options={})
|
|
host = "0.0.0.0"
|
|
port = options[:port]
|
|
trust_port = port + 1
|
|
connect_time_stale_client = options[:connect_time_stale_client].to_i
|
|
connect_time_expire_client = options[:connect_time_expire_client].to_i
|
|
connect_time_stale_browser = options[:connect_time_stale_browser].to_i
|
|
connect_time_expire_browser = options[:connect_time_expire_browser].to_i
|
|
max_connections_per_user = options[:max_connections_per_user].to_i
|
|
gateway_name = options[:gateway_name]
|
|
rabbitmq_host = options[:rabbitmq_host]
|
|
rabbitmq_port = options[:rabbitmq_port].to_i
|
|
calling_thread = options[:calling_thread]
|
|
trust_check = TrustCheck.new(trust_port, options[:cidr])
|
|
|
|
@log.info "starting server #{host}:#{port} staleness_time=#{connect_time_stale_client}; reconnect time = #{connect_time_expire_client}, rabbitmq=#{rabbitmq_host}:#{rabbitmq_port}"
|
|
|
|
EventMachine.error_handler{|e|
|
|
@log.error "unhandled error #{e}"
|
|
Bugsnag.notify(e)
|
|
}
|
|
|
|
EventMachine.run do
|
|
@router.start(connect_time_stale_client, connect_time_expire_client, connect_time_stale_browser, connect_time_expire_browser, host: rabbitmq_host, port: rabbitmq_port, max_connections_per_user: max_connections_per_user, gateway: gateway_name) do
|
|
start_connection_expiration
|
|
start_client_expiration
|
|
start_connection_flagger
|
|
start_stats_dump
|
|
start_websocket_listener(host, port, trust_port, trust_check, options[:emwebsocket_debug])
|
|
calling_thread.wakeup if calling_thread
|
|
end
|
|
|
|
# if you don't do this, the app won't exit unless you kill -9
|
|
at_exit do
|
|
@log.info "cleaning up server"
|
|
@router.cleanup
|
|
end
|
|
end
|
|
end
|
|
|
|
def stop
|
|
EventMachine::stop_event_loop
|
|
end
|
|
|
|
def start_websocket_listener(listen_ip, port, trust_port, trust_check, emwebsocket_debug)
|
|
EventMachine::WebSocket.run(:host => listen_ip, :port => port, :debug => emwebsocket_debug) do |ws|
|
|
#@log.info "new client #{ws}"
|
|
@router.new_client(ws, false)
|
|
end
|
|
EventMachine::WebSocket.run(:host => listen_ip, :port => trust_port, :debug => emwebsocket_debug) do |ws|
|
|
@log.info "new latency_tester client #{ws}"
|
|
# verify this connection came in from a valid subnet, if specified
|
|
ip = extract_ip(ws)
|
|
if trust_check.trusted?(ip, trust_port)
|
|
@router.new_client(ws, true)
|
|
else
|
|
@log.warn("untrusted client attempted to connect to #{listen_ip}:#{trust_port} from #{ip}")
|
|
ws.close
|
|
end
|
|
end
|
|
@log.debug("started websocket")
|
|
end
|
|
|
|
def start_connection_expiration
|
|
# one cleanup on startup
|
|
@router.periodical_check_connections
|
|
|
|
EventMachine::PeriodicTimer.new(2) do
|
|
sane_logging { @router.periodical_check_connections }
|
|
end
|
|
end
|
|
|
|
def start_client_expiration
|
|
# one cleanup on startup
|
|
@router.periodical_check_clients
|
|
|
|
EventMachine::PeriodicTimer.new(30) do
|
|
sane_logging { @router.periodical_check_clients }
|
|
end
|
|
end
|
|
|
|
def start_connection_flagger
|
|
# one cleanup on startup
|
|
@router.periodical_flag_connections
|
|
|
|
EventMachine::PeriodicTimer.new(2) do
|
|
sane_logging { @router.periodical_flag_connections }
|
|
end
|
|
end
|
|
|
|
def start_stats_dump
|
|
EventMachine::PeriodicTimer.new(60) do
|
|
@router.periodical_stats_dump
|
|
end
|
|
end
|
|
|
|
def sane_logging(&blk)
|
|
# used around repeated transactions that cause too much ActiveRecord::Base logging
|
|
# example is handling heartbeats
|
|
begin
|
|
original_level = @ar_base_logger.level if @ar_base_logger
|
|
@ar_base_logger.level = :info if @ar_base_logger
|
|
blk.call
|
|
ensure
|
|
@ar_base_logger.level = original_level if @ar_base_logger
|
|
end
|
|
end
|
|
|
|
private
|
|
def extract_ip(client)
|
|
Socket.unpack_sockaddr_in(client.get_peername)[1]
|
|
end
|
|
end
|
|
|
|
end
|