trabajo-afectivo/lib/sessions.rb

638 lines
12 KiB
Ruby
Raw Normal View History

# Copyright (C) 2012-2021 Zammad Foundation, http://zammad-foundation.org/
module Sessions
2013-01-24 01:01:47 +00:00
@store = case Rails.application.config.websocket_session_store
when :redis then Sessions::Store::Redis.new
else Sessions::Store::File.new
end
2013-01-24 01:01:47 +00:00
# create global vars for threads
@@client_threads = {} # rubocop:disable Style/ClassVars
2012-07-23 22:22:23 +00:00
2014-06-27 06:43:37 +00:00
=begin
start new session
Sessions.create(client_id, session_data, { type: 'websocket' })
2014-06-27 06:43:37 +00:00
returns
true|false
=end
def self.create(client_id, session, meta)
# collect session data
2015-05-10 20:53:15 +00:00
meta[:last_ping] = Time.now.utc.to_i
data = {
user: session,
meta: meta,
}
content = data.to_json
@store.create(client_id, content)
2012-11-02 16:10:22 +00:00
# send update to browser
2017-11-23 08:09:44 +00:00
return if !session || session['id'].blank?
2017-11-23 08:09:44 +00:00
send(
client_id,
{
event: 'ws:login',
data: { success: true },
2017-11-23 08:09:44 +00:00
}
)
2012-07-23 22:22:23 +00:00
end
2014-06-27 06:43:37 +00:00
=begin
list of all session
client_ids = Sessions.sessions
returns
['4711', '4712']
=end
def self.sessions
@store.sessions
2014-06-27 06:43:37 +00:00
end
=begin
list of all session
Sessions.session_exists?(client_id)
returns
true|false
=end
def self.session_exists?(client_id)
@store.session_exists?(client_id)
2014-06-27 06:43:37 +00:00
end
=begin
list of all session with data
client_ids_with_data = Sessions.list
returns
{
'4711' => {
user: {
'id' => 123,
2014-06-27 06:43:37 +00:00
},
meta: {
type: 'websocket',
last_ping: time_of_last_ping,
2014-06-27 06:43:37 +00:00
}
},
'4712' => {
user: {
'id' => 124,
2014-06-27 06:43:37 +00:00
},
meta: {
type: 'ajax',
last_ping: time_of_last_ping,
2014-06-27 06:43:37 +00:00
}
},
}
=end
def self.list
client_ids = sessions
2014-06-27 06:43:37 +00:00
session_list = {}
client_ids.each do |client_id|
data = get(client_id)
2014-06-27 06:43:37 +00:00
next if !data
2014-06-27 06:43:37 +00:00
session_list[client_id] = data
end
2014-06-27 06:43:37 +00:00
session_list
end
=begin
destroy session
Sessions.destroy(client_id)
2014-06-27 06:43:37 +00:00
returns
true|false
=end
def self.destroy(client_id)
@store.destroy(client_id)
2014-06-27 06:43:37 +00:00
end
=begin
destroy idle session
list_of_client_ids = Sessions.destroy_idle_sessions
2014-06-27 06:43:37 +00:00
returns
['4711', '4712']
=end
def self.destroy_idle_sessions(idle_time_in_sec = 240)
2014-06-27 06:43:37 +00:00
list_of_closed_sessions = []
clients = Sessions.list
clients.each do |client_id, client|
if !client[:meta] || !client[:meta][:last_ping] || (client[:meta][:last_ping].to_i + idle_time_in_sec) < Time.now.utc.to_i
2014-06-27 06:43:37 +00:00
list_of_closed_sessions.push client_id
Sessions.destroy(client_id)
2014-06-27 06:43:37 +00:00
end
end
2014-06-27 06:43:37 +00:00
list_of_closed_sessions
end
=begin
touch session
Sessions.touch(client_id)
returns
true|false
=end
def self.touch(client_id)
data = get(client_id)
2014-06-27 06:43:37 +00:00
return false if !data
2015-05-10 20:53:15 +00:00
data[:meta][:last_ping] = Time.now.utc.to_i
@store.set(client_id, data)
2014-06-27 06:43:37 +00:00
true
end
=begin
get session data
data = Sessions.get(client_id)
returns
{
user: {
'id' => 123,
2014-06-27 06:43:37 +00:00
},
meta: {
type: 'websocket',
last_ping: time_of_last_ping,
2014-06-27 06:43:37 +00:00
}
}
=end
def self.get(client_id)
@store.get client_id
2014-06-27 06:43:37 +00:00
end
=begin
send message to client
Sessions.send(client_id_of_recipient, data)
returns
true|false
=end
def self.send(client_id, data)
@store.send_data(client_id, data)
2014-06-27 06:43:37 +00:00
end
=begin
send message to recipient client
Sessions.send_to(user_id, data)
e. g.
Sessions.send_to(user_id, {
event: 'session:takeover',
data: {
taskbar_id: 12312
},
})
returns
true|false
=end
def self.send_to(user_id, data)
# list all current clients
client_list = sessions
client_list.each do |client_id|
session = Sessions.get(client_id)
next if !session
next if !session[:user]
next if !session[:user]['id']
next if session[:user]['id'].to_i != user_id.to_i
Sessions.send(client_id, data)
end
true
end
=begin
send message to all authenticated client
2014-06-27 06:43:37 +00:00
Sessions.broadcast(data)
returns
2016-05-26 08:14:51 +00:00
[array_with_client_ids_of_recipients]
2014-06-27 06:43:37 +00:00
broadcase also to not authenticated client
Sessions.broadcast(data, 'public') # public|authenticated
broadcase also not to sender
Sessions.broadcast(data, 'public', sender_user_id)
2014-06-27 06:43:37 +00:00
=end
def self.broadcast(data, recipient = 'authenticated', sender_user_id = nil)
2014-06-27 06:43:37 +00:00
# list all current clients
2016-05-26 08:14:51 +00:00
recipients = []
client_list = sessions
client_list.each do |client_id|
session = Sessions.get(client_id)
next if !session
if recipient != 'public'
next if session[:user].blank?
next if session[:user]['id'].blank?
end
next if sender_user_id && session[:user] && session[:user]['id'] && session[:user]['id'].to_i == sender_user_id.to_i
Sessions.send(client_id, data)
2016-05-26 08:14:51 +00:00
recipients.push client_id
end
2016-05-26 08:14:51 +00:00
recipients
2014-06-27 06:43:37 +00:00
end
=begin
get messages for client
messages = Sessions.queue(client_id_of_recipient)
returns
[
{
key1 => 'some data of message 1',
key2 => 'some data of message 1',
},
{
key1 => 'some data of message 2',
key2 => 'some data of message 2',
},
]
=end
def self.queue(client_id)
@store.queue(client_id)
2014-06-27 06:43:37 +00:00
end
2017-05-02 17:16:46 +00:00
=begin
remove all session and spool messages
Sessions.cleanup
=end
def self.cleanup
@store.cleanup
2014-06-29 19:30:55 +00:00
end
=begin
create spool messages
Sessions.spool_create(some: 'data')
=end
2015-12-09 13:09:37 +00:00
def self.spool_create(data)
msg = JSON.generate(data)
data = {
msg: msg,
timestamp: Time.now.utc.to_i,
}
@store.add_to_spool(data)
end
=begin
get spool messages
Sessions.spool_list(junger_then, for_user_id)
=end
def self.spool_list(timestamp, current_user_id)
data = []
to_delete = []
@store.each_spool do |message, entry|
message_parsed = {}
begin
spool = JSON.parse(message)
message_parsed = JSON.parse(spool['msg'])
rescue => e
log('error', "can't parse spool message: #{message}, #{e.inspect}")
to_delete.push [message, entry]
next
end
# ignore message older then 48h
if spool['timestamp'] + (2 * 86_400) < Time.now.utc.to_i
to_delete.push [message, entry]
next
end
# add spool attribute to push spool info to clients
message_parsed['spool'] = true
# only send not already older messages
if !timestamp || timestamp < spool['timestamp']
# spool to recipient list
if message_parsed['recipient'] && message_parsed['recipient']['user_id']
message_parsed['recipient']['user_id'].each do |user_id|
2015-12-09 13:09:37 +00:00
next if current_user_id != user_id
2015-12-09 13:09:37 +00:00
message = message_parsed
if message_parsed['event'] == 'broadcast'
message = message_parsed['data']
end
item = {
type: 'direct',
2015-12-09 13:09:37 +00:00
message: message,
}
data.push item
end
# spool to every client
else
message = message_parsed
if message_parsed['event'] == 'broadcast'
message = message_parsed['data']
end
item = {
type: 'broadcast',
message: message,
}
data.push item
end
end
end
to_delete.each do |item|
@store.remove_from_spool(*item)
end
data
end
=begin
delete spool messages
Sessions.spool_delete
=end
def self.spool_delete
@store.clear_spool
end
def self.jobs(node_id = nil)
# dispatch sessions
if node_id.blank? && ENV['ZAMMAD_SESSION_JOBS_CONCURRENT'].to_i.positive?
previous_nodes_sessions = Sessions::Node.stats
if previous_nodes_sessions.present?
log('info', "Cleaning up previous Sessions::Node sessions: #{previous_nodes_sessions}")
Sessions::Node.cleanup
end
dispatcher_pid = Process.pid
node_count = ENV['ZAMMAD_SESSION_JOBS_CONCURRENT'].to_i
node_pids = []
(1..node_count).each do |worker_node_id|
node_pids << fork do
title = "Zammad Session Jobs Node ##{worker_node_id}: dispatch_pid:#{dispatcher_pid} -> worker_pid:#{Process.pid}"
$PROGRAM_NAME = title
log('info', "#{title} started.")
::Sessions.jobs(worker_node_id)
sleep node_count
rescue Interrupt
nil
end
end
Signal.trap 'SIGTERM' do
node_pids.each do |node_pid|
Process.kill 'TERM', node_pid
end
Process.waitall
raise SignalException, 'SIGTERM'
end
# dispatch client_ids to nodes
loop do
# nodes
nodes_stats = Sessions::Node.stats
client_ids = sessions
client_ids.each do |client_id|
# ask nodes for nodes
next if nodes_stats[client_id]
# assign to node
Sessions::Node.session_assigne(client_id)
2018-01-24 12:55:15 +00:00
sleep 1
end
sleep 1
end
end
2012-08-03 22:46:05 +00:00
Thread.abort_on_exception = true
loop do
if node_id
# register node
Sessions::Node.register(node_id)
# watch for assigned sessions
client_ids = Sessions::Node.sessions_by(node_id)
else
client_ids = sessions
end
client_ids.each do |client_id|
2012-07-23 22:22:23 +00:00
2014-06-29 19:30:55 +00:00
# connection already open, ignore
2012-08-07 21:53:43 +00:00
next if @@client_threads[client_id]
# get current user
session_data = Sessions.get(client_id)
2018-01-24 12:55:15 +00:00
next if session_data.blank?
next if session_data[:user].blank?
next if session_data[:user]['id'].blank?
2017-11-06 01:32:47 +00:00
user = User.lookup(id: session_data[:user]['id'])
2018-01-24 12:55:15 +00:00
next if user.blank?
2012-08-03 22:46:05 +00:00
# start client thread
2018-01-24 12:55:15 +00:00
next if @@client_threads[client_id].present?
@@client_threads[client_id] = true
@@client_threads[client_id] = Thread.new do
thread_client(client_id, 0, Time.now.utc, node_id)
@@client_threads[client_id] = nil
log('debug', "close client (#{client_id}) thread")
2017-11-06 01:32:47 +00:00
if ActiveRecord::Base.connection.owner == Thread.current
ActiveRecord::Base.connection.close
end
end
2018-01-24 12:55:15 +00:00
sleep 1
end
2012-08-07 06:43:15 +00:00
2018-01-24 12:55:15 +00:00
sleep 1
2012-07-23 22:22:23 +00:00
end
end
2014-06-29 19:30:55 +00:00
=begin
check if thread for client_id is running
Sessions.thread_client_exists?(client_id)
returns
thread
=end
def self.thread_client_exists?(client_id)
@@client_threads[client_id]
2013-12-01 11:55:21 +00:00
end
2014-06-29 19:30:55 +00:00
=begin
start client for browser
Sessions.thread_client(client_id)
returns
thread
=end
def self.thread_client(client_id, try_count = 0, try_run_time = Time.now.utc, node_id)
log('debug', "LOOP #{node_id}.#{client_id} - #{try_count}")
2014-06-29 19:30:55 +00:00
begin
Sessions::Client.new(client_id, node_id)
2014-06-29 19:30:55 +00:00
rescue => e
log('error', "thread_client #{client_id} exited with error #{e.inspect}")
log('error', e.backtrace.join("\n "))
2014-06-29 19:30:55 +00:00
sleep 10
2013-12-01 11:55:21 +00:00
begin
2014-06-29 19:30:55 +00:00
ActiveRecord::Base.connection_pool.release_connection
2013-12-01 11:55:21 +00:00
rescue => e
log('error', "Can't reconnect to database #{e.inspect}")
2014-06-29 19:30:55 +00:00
end
2014-04-24 05:49:30 +00:00
2014-06-29 19:30:55 +00:00
try_run_max = 10
try_count += 1
2014-04-24 05:49:30 +00:00
2014-06-29 19:30:55 +00:00
# reset error counter if to old
if try_run_time + (60 * 5) < Time.now.utc
2014-06-29 19:30:55 +00:00
try_count = 0
end
2015-05-10 19:47:17 +00:00
try_run_time = Time.now.utc
2014-04-24 05:49:30 +00:00
2014-06-29 19:30:55 +00:00
# restart job again
if try_run_max > try_count
thread_client(client_id, try_count, try_run_time, node_id)
2013-12-01 11:55:21 +00:00
end
raise "STOP thread_client for client #{node_id}.#{client_id} after #{try_run_max} tries"
2014-06-29 19:30:55 +00:00
end
log('debug', "/LOOP #{node_id}.#{client_id} - #{try_count}")
2013-12-01 11:55:21 +00:00
end
def self.symbolize_keys(hash)
hash.each_with_object({}) do |(key, value), result|
new_key = case key
when String then key.to_sym
else key
end
new_value = case value
when Hash then symbolize_keys(value)
else value
end
result[new_key] = new_value
end
end
# we use it in rails and non rails context
def self.log(level, message)
if defined?(Rails)
case level
when 'debug'
Rails.logger.debug { message }
when 'info'
Rails.logger.info message
else
Rails.logger.error message
end
return
end
puts "#{Time.now.utc.iso8601}:#{level} #{message}" # rubocop:disable Rails/Output
end
end