trabajo-afectivo/lib/sessions.rb

778 lines
15 KiB
Ruby
Raw Normal View History

module Sessions
2013-01-24 01:01:47 +00:00
# get application root directory
@root = Dir.pwd.to_s
if @root.blank? || @root == '/'
2013-01-24 01:01:47 +00:00
@root = Rails.root
end
# get working directories
@path = "#{@root}/tmp/websocket_#{Rails.env}"
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)
path = "#{@path}/#{client_id}"
path_tmp = "#{@path}/tmp/#{client_id}"
session_file = "#{path_tmp}/session"
# 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 session data in session file
FileUtils.mkpath path_tmp
File.open(session_file, 'wb') do |file|
file.write content
end
2012-11-02 16:10:22 +00:00
# destroy old session if needed
if File.exist?(path)
Sessions.destroy(client_id)
2015-05-03 12:11:47 +00:00
end
# move to destination directory
FileUtils.mv(path_tmp, path)
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
path = "#{@path}/"
2014-06-27 06:43:37 +00:00
# just make sure that spool path exists
if !File.exist?(path)
2014-06-27 06:43:37 +00:00
FileUtils.mkpath path
end
data = []
Dir.foreach(path) do |entry|
next if entry == '.'
next if entry == '..'
next if entry == 'tmp'
next if entry == 'spool'
2014-06-27 06:43:37 +00:00
data.push entry.to_s
end
data
end
=begin
list of all session
Sessions.session_exists?(client_id)
returns
true|false
=end
def self.session_exists?(client_id)
session_dir = "#{@path}/#{client_id}"
return false if !File.exist?(session_dir)
session_file = "#{session_dir}/session"
return false if !File.exist?(session_file)
true
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)
path = "#{@path}/#{client_id}"
2014-06-27 06:43:37 +00:00
FileUtils.rm_rf path
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|
2015-05-10 19:47:17 +00:00
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
path = "#{@path}/#{client_id}"
2015-05-10 20:53:15 +00:00
data[:meta][:last_ping] = Time.now.utc.to_i
File.open("#{path}/session", 'wb' ) do |file|
file.flock(File::LOCK_EX)
file.write data.to_json
file.flock(File::LOCK_UN)
end
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)
session_dir = "#{@path}/#{client_id}"
session_file = "#{session_dir}/session"
data = nil
2015-05-12 08:46:07 +00:00
# if no session dir exists, session got destoried
if !File.exist?(session_dir)
destroy(client_id)
log('debug', "missing session directory #{session_dir} for '#{client_id}', remove session.")
2015-05-03 12:11:47 +00:00
return
end
2015-05-12 08:46:07 +00:00
# if only session file is missing, then it's an error behavior
if !File.exist?(session_file)
destroy(client_id)
log('error', "missing session file for '#{client_id}', remove session.")
2014-10-23 18:08:00 +00:00
return
end
2014-06-27 06:43:37 +00:00
begin
File.open(session_file, 'rb') do |file|
file.flock(File::LOCK_SH)
2014-06-27 06:43:37 +00:00
all = file.read
file.flock(File::LOCK_UN)
data_json = JSON.parse(all)
if data_json
2015-05-10 20:53:15 +00:00
data = symbolize_keys(data_json)
data[:user] = data_json['user'] # for compat. reasons
end
end
rescue => e
log('error', e.inspect)
destroy(client_id)
log('error', "error in reading/parsing session file '#{session_file}', remove session.")
2014-06-27 06:43:37 +00:00
return
end
data
end
=begin
send message to client
Sessions.send(client_id_of_recipient, data)
returns
true|false
=end
def self.send(client_id, data)
path = "#{@path}/#{client_id}/"
filename = "send-#{Time.now.utc.to_f}"
location = "#{path}#{filename}"
check = true
count = 0
2014-06-27 06:43:37 +00:00
while check
if File.exist?(location)
2014-06-27 06:43:37 +00:00
count += 1
location = "#{path}#{filename}-#{count}"
2014-06-27 06:43:37 +00:00
else
check = false
end
end
return false if !File.directory? path
begin
File.open(location, 'wb') do |file|
file.flock(File::LOCK_EX)
file.write data.to_json
file.flock(File::LOCK_UN)
file.close
end
rescue => e
log('error', e.inspect)
log('error', "error in writing message file '#{location}'")
return false
end
2014-06-27 06:43:37 +00:00
true
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
if sender_user_id
next if session[:user] && session[:user]['id'] && session[:user]['id'].to_i == sender_user_id.to_i
end
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)
path = "#{@path}/#{client_id}/"
data = []
2014-06-27 06:43:37 +00:00
files = []
Dir.foreach(path) do |entry|
next if entry == '.'
next if entry == '..'
2014-06-27 06:43:37 +00:00
files.push entry
end
files.sort.each do |entry|
next if entry !~ /^send/
message = Sessions.queue_file_read(path, entry)
next if !message
data.push message
end
2014-06-27 06:43:37 +00:00
data
end
def self.queue_file_read(path, filename)
location = "#{path}#{filename}"
message = ''
File.open(location, 'rb') do |file|
file.flock(File::LOCK_EX)
message = file.read
file.flock(File::LOCK_UN)
end
File.delete(location)
return if message.blank?
begin
return JSON.parse(message)
rescue => e
log('error', "can't parse queue message: #{message}, #{e.inspect}")
return
end
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
2017-05-02 17:41:35 +00:00
return true if !File.exist?(@path)
2017-05-02 17:41:35 +00:00
FileUtils.rm_rf @path
2017-05-02 17:16:46 +00:00
true
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)
path = "#{@path}/spool/"
FileUtils.mkpath path
data = {
msg: msg,
timestamp: Time.now.utc.to_i,
}
2015-12-09 13:09:37 +00:00
file_path = "#{path}/#{Time.now.utc.to_f}-#{rand(99_999)}"
File.open(file_path, 'wb') do |file|
file.flock(File::LOCK_EX)
file.write data.to_json
file.flock(File::LOCK_UN)
end
end
=begin
get spool messages
Sessions.spool_list(junger_then, for_user_id)
=end
def self.spool_list(timestamp, current_user_id)
path = "#{@path}/spool/"
FileUtils.mkpath path
data = []
to_delete = []
files = []
Dir.foreach(path) do |entry|
next if entry == '.'
next if entry == '..'
files.push entry
end
files.sort.each do |entry|
filename = "#{path}/#{entry}"
next if !File.exist?(filename)
File.open(filename, 'rb') do |file|
file.flock(File::LOCK_SH)
message = file.read
file.flock(File::LOCK_UN)
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 "#{path}/#{entry}"
next
end
# ignore message older then 48h
2015-05-10 19:47:17 +00:00
if spool['timestamp'] + (2 * 86_400) < Time.now.utc.to_i
to_delete.push "#{path}/#{entry}"
next
end
# add spool attribute to push spool info to clients
2013-06-28 22:26:04 +00:00
message_parsed['spool'] = true
# only send not already older messages
if !timestamp || timestamp < spool['timestamp']
# spool to recipient list
2013-06-28 22:26:04 +00:00
if message_parsed['recipient'] && message_parsed['recipient']['user_id']
message_parsed['recipient']['user_id'].each do |user_id|
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
2015-12-09 13:09:37 +00:00
message = message_parsed
if message_parsed['event'] == 'broadcast'
message = message_parsed['data']
end
item = {
type: 'broadcast',
2015-12-09 13:09:37 +00:00
message: message,
}
data.push item
end
end
end
end
to_delete.each do |file|
File.delete(file)
end
data
end
=begin
delete spool messages
Sessions.spool_delete
=end
def self.spool_delete
path = "#{@path}/spool/"
FileUtils.rm_rf path
end
def self.jobs(node_id = nil)
# just make sure that spool path exists
if !File.exist?(@path)
2013-01-15 23:10:27 +00:00
FileUtils.mkpath @path
end
# dispatch sessions
2017-12-04 07:05:31 +00:00
if node_id&.zero?
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]
# assigne 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
2015-05-10 19:47:17 +00:00
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)
if level == 'debug'
Rails.logger.debug { message }
elsif level == 'notice'
Rails.logger.notice message
else
Rails.logger.error message
end
return
end
puts "#{Time.now.utc.iso8601}:#{level} #{message}" # rubocop:disable Rails/Output
end
end