sessions.rb 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550
  1. # Copyright (C) 2012-2024 Zammad Foundation, https://zammad-foundation.org/
  2. module Sessions
  3. @store = case Rails.application.config.websocket_session_store
  4. when :redis then Sessions::Store::Redis.new
  5. else Sessions::Store::File.new
  6. end
  7. # create global vars for threads
  8. @@client_threads = {} # rubocop:disable Style/ClassVars
  9. =begin
  10. start new session
  11. Sessions.create(client_id, session_data, { type: 'websocket' })
  12. returns
  13. true|false
  14. =end
  15. def self.create(client_id, session, meta)
  16. # collect session data
  17. meta[:last_ping] = Time.now.utc.to_i
  18. data = {
  19. user: session,
  20. meta: meta,
  21. }
  22. content = data.to_json
  23. @store.create(client_id, content)
  24. # send update to browser
  25. return if !session || session['id'].blank?
  26. send(
  27. client_id,
  28. {
  29. event: 'ws:login',
  30. data: { success: true },
  31. }
  32. )
  33. end
  34. =begin
  35. list of all session
  36. client_ids = Sessions.sessions
  37. returns
  38. ['4711', '4712']
  39. =end
  40. def self.sessions
  41. @store.sessions
  42. end
  43. =begin
  44. list of all session
  45. Sessions.session_exists?(client_id)
  46. returns
  47. true|false
  48. =end
  49. def self.session_exists?(client_id)
  50. @store.session_exists?(client_id)
  51. end
  52. =begin
  53. list of all session with data
  54. client_ids_with_data = Sessions.list
  55. returns
  56. {
  57. '4711' => {
  58. user: {
  59. 'id' => 123,
  60. },
  61. meta: {
  62. type: 'websocket',
  63. last_ping: time_of_last_ping,
  64. }
  65. },
  66. '4712' => {
  67. user: {
  68. 'id' => 124,
  69. },
  70. meta: {
  71. type: 'ajax',
  72. last_ping: time_of_last_ping,
  73. }
  74. },
  75. }
  76. =end
  77. def self.list
  78. client_ids = sessions
  79. session_list = {}
  80. client_ids.each do |client_id|
  81. data = get(client_id)
  82. next if !data
  83. session_list[client_id] = data
  84. end
  85. session_list
  86. end
  87. =begin
  88. destroy session
  89. Sessions.destroy(client_id)
  90. returns
  91. true|false
  92. =end
  93. def self.destroy(client_id)
  94. @store.destroy(client_id)
  95. end
  96. =begin
  97. destroy idle session
  98. list_of_client_ids = Sessions.destroy_idle_sessions
  99. returns
  100. ['4711', '4712']
  101. =end
  102. def self.destroy_idle_sessions(idle_time_in_sec = 240)
  103. list_of_closed_sessions = []
  104. clients = Sessions.list
  105. clients.each do |client_id, client|
  106. if !client[:meta] || !client[:meta][:last_ping] || (client[:meta][:last_ping].to_i + idle_time_in_sec) < Time.now.utc.to_i
  107. list_of_closed_sessions.push client_id
  108. Sessions.destroy(client_id)
  109. end
  110. end
  111. list_of_closed_sessions
  112. end
  113. =begin
  114. touch session
  115. Sessions.touch(client_id)
  116. returns
  117. true|false
  118. =end
  119. def self.touch(client_id)
  120. data = get(client_id)
  121. return false if !data
  122. data[:meta][:last_ping] = Time.now.utc.to_i
  123. @store.set(client_id, data)
  124. true
  125. end
  126. =begin
  127. get session data
  128. data = Sessions.get(client_id)
  129. returns
  130. {
  131. user: {
  132. 'id' => 123,
  133. },
  134. meta: {
  135. type: 'websocket',
  136. last_ping: time_of_last_ping,
  137. }
  138. }
  139. =end
  140. def self.get(client_id)
  141. @store.get client_id
  142. end
  143. =begin
  144. send message to client
  145. Sessions.send(client_id_of_recipient, data)
  146. returns
  147. true|false
  148. =end
  149. def self.send(client_id, data) # rubocop:disable Zammad/ForbidDefSend
  150. @store.send_data(client_id, data)
  151. end
  152. =begin
  153. send message to recipient client
  154. Sessions.send_to(user_id, data)
  155. e. g.
  156. Sessions.send_to(user_id, {
  157. event: 'session_takeover',
  158. data: {
  159. taskbar_id: 12312
  160. },
  161. })
  162. returns
  163. true|false
  164. =end
  165. def self.send_to(user_id, data)
  166. # list all current clients
  167. client_list = sessions
  168. client_list.each do |client_id|
  169. session = Sessions.get(client_id)
  170. next if !session
  171. next if !session[:user]
  172. next if !session[:user]['id']
  173. next if session[:user]['id'].to_i != user_id.to_i
  174. Sessions.send(client_id, data)
  175. end
  176. true
  177. end
  178. =begin
  179. send message to all authenticated client
  180. Sessions.broadcast(data)
  181. returns
  182. [array_with_client_ids_of_recipients]
  183. broadcase also to not authenticated client
  184. Sessions.broadcast(data, 'public') # public|authenticated
  185. broadcase also not to sender
  186. Sessions.broadcast(data, 'public', sender_user_id)
  187. =end
  188. def self.broadcast(data, recipient = 'authenticated', sender_user_id = nil)
  189. # list all current clients
  190. recipients = []
  191. client_list = sessions
  192. client_list.each do |client_id|
  193. session = Sessions.get(client_id)
  194. next if !session
  195. if recipient != 'public'
  196. next if session[:user].blank?
  197. next if session[:user]['id'].blank?
  198. end
  199. next if sender_user_id && session[:user] && session[:user]['id'] && session[:user]['id'].to_i == sender_user_id.to_i
  200. Sessions.send(client_id, data)
  201. recipients.push client_id
  202. end
  203. recipients
  204. end
  205. =begin
  206. get messages for client
  207. messages = Sessions.queue(client_id_of_recipient)
  208. returns
  209. [
  210. {
  211. key1 => 'some data of message 1',
  212. key2 => 'some data of message 1',
  213. },
  214. {
  215. key1 => 'some data of message 2',
  216. key2 => 'some data of message 2',
  217. },
  218. ]
  219. =end
  220. def self.queue(client_id)
  221. @store.queue(client_id)
  222. end
  223. =begin
  224. remove all session and spool messages
  225. Sessions.cleanup
  226. =end
  227. def self.cleanup
  228. @store.cleanup
  229. end
  230. =begin
  231. Zammad previously used a spooling mechanism for session mechanism.
  232. The code to clean-up such data is still here, even though the mechanism
  233. itself was removed in the meantime.
  234. Sessions.spool_delete
  235. =end
  236. def self.spool_delete
  237. @store.clear_spool
  238. end
  239. def self.jobs(node_id = nil)
  240. # dispatch sessions
  241. if node_id.blank? && ENV['ZAMMAD_SESSION_JOBS_CONCURRENT'].to_i.positive?
  242. previous_nodes_sessions = Sessions::Node.stats
  243. if previous_nodes_sessions.present?
  244. log('info', "Cleaning up previous Sessions::Node sessions: #{previous_nodes_sessions}")
  245. Sessions::Node.cleanup
  246. end
  247. dispatcher_pid = Process.pid
  248. node_count = ENV['ZAMMAD_SESSION_JOBS_CONCURRENT'].to_i
  249. node_pids = (1..node_count).map do |worker_node_id|
  250. fork do
  251. title = "Zammad Session Jobs Node ##{worker_node_id}: dispatch_pid:#{dispatcher_pid} -> worker_pid:#{Process.pid}"
  252. $PROGRAM_NAME = title
  253. log('info', "#{title} started.")
  254. ::Sessions.jobs(worker_node_id)
  255. sleep node_count
  256. rescue Interrupt
  257. nil
  258. end
  259. end
  260. Signal.trap 'SIGTERM' do
  261. node_pids.each do |node_pid|
  262. Process.kill 'TERM', node_pid
  263. end
  264. Process.waitall
  265. raise SignalException, 'SIGTERM'
  266. end
  267. # dispatch client_ids to nodes
  268. loop do
  269. # nodes
  270. nodes_stats = Sessions::Node.stats
  271. client_ids = sessions
  272. client_ids.each do |client_id|
  273. # ask nodes for nodes
  274. next if nodes_stats[client_id]
  275. # assign to node
  276. Sessions::Node.session_assigne(client_id)
  277. sleep 1
  278. end
  279. sleep 1
  280. end
  281. end
  282. Thread.abort_on_exception = true
  283. loop do
  284. if node_id
  285. # register node
  286. Sessions::Node.register(node_id)
  287. # watch for assigned sessions
  288. client_ids = Sessions::Node.sessions_by(node_id)
  289. else
  290. client_ids = sessions
  291. end
  292. client_ids.each do |client_id|
  293. # connection already open, ignore
  294. next if @@client_threads[client_id]
  295. # get current user
  296. session_data = Sessions.get(client_id)
  297. next if session_data.blank?
  298. next if session_data[:user].blank?
  299. next if session_data[:user]['id'].blank?
  300. user = User.lookup(id: session_data[:user]['id'])
  301. next if user.blank?
  302. # start client thread
  303. next if @@client_threads[client_id].present?
  304. @@client_threads[client_id] = true
  305. @@client_threads[client_id] = Thread.new do
  306. thread_client(client_id, 0, Time.now.utc, node_id)
  307. @@client_threads[client_id] = nil
  308. log('debug', "close client (#{client_id}) thread")
  309. if ActiveRecord::Base.connection.owner == Thread.current
  310. ActiveRecord::Base.connection.close
  311. end
  312. end
  313. sleep 1
  314. end
  315. sleep 1
  316. end
  317. end
  318. =begin
  319. check if thread for client_id is running
  320. Sessions.thread_client_exists?(client_id)
  321. returns
  322. thread
  323. =end
  324. def self.thread_client_exists?(client_id)
  325. @@client_threads[client_id]
  326. end
  327. =begin
  328. start client for browser
  329. Sessions.thread_client(client_id)
  330. returns
  331. thread
  332. =end
  333. def self.thread_client(client_id, try_count = 0, try_run_time = Time.now.utc, node_id)
  334. log('debug', "LOOP #{node_id}.#{client_id} - #{try_count}")
  335. begin
  336. Sessions::Client.new(client_id, node_id)
  337. rescue => e
  338. log('error', "thread_client #{client_id} exited with error #{e.inspect}")
  339. log('error', e.backtrace.join("\n "))
  340. sleep 10
  341. begin
  342. ActiveRecord::Base.connection_pool.release_connection
  343. rescue => e
  344. log('error', "Can't reconnect to database #{e.inspect}")
  345. end
  346. try_run_max = 10
  347. try_count += 1
  348. # reset error counter if to old
  349. if try_run_time + (60 * 5) < Time.now.utc
  350. try_count = 0
  351. end
  352. try_run_time = Time.now.utc
  353. # restart job again
  354. if try_run_max > try_count
  355. thread_client(client_id, try_count, try_run_time, node_id)
  356. end
  357. raise "STOP thread_client for client #{node_id}.#{client_id} after #{try_run_max} tries"
  358. ensure
  359. ActiveSupport::CurrentAttributes.clear_all
  360. end
  361. log('debug', "/LOOP #{node_id}.#{client_id} - #{try_count}")
  362. end
  363. def self.symbolize_keys(hash)
  364. hash.each_with_object({}) do |(key, value), result|
  365. new_key = case key
  366. when String then key.to_sym
  367. else key
  368. end
  369. new_value = case value
  370. when Hash then symbolize_keys(value)
  371. else value
  372. end
  373. result[new_key] = new_value
  374. end
  375. end
  376. # we use it in rails and non rails context
  377. def self.log(level, message)
  378. if defined?(Rails)
  379. case level
  380. when 'debug'
  381. Rails.logger.debug { message }
  382. when 'info'
  383. Rails.logger.info message
  384. else
  385. Rails.logger.error message
  386. end
  387. return
  388. end
  389. puts "#{Time.now.utc.iso8601}:#{level} #{message}" # rubocop:disable Rails/Output
  390. end
  391. end