sessions.rb 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551
  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 = []
  250. (1..node_count).each do |worker_node_id|
  251. node_pids << fork do
  252. title = "Zammad Session Jobs Node ##{worker_node_id}: dispatch_pid:#{dispatcher_pid} -> worker_pid:#{Process.pid}"
  253. $PROGRAM_NAME = title
  254. log('info', "#{title} started.")
  255. ::Sessions.jobs(worker_node_id)
  256. sleep node_count
  257. rescue Interrupt
  258. nil
  259. end
  260. end
  261. Signal.trap 'SIGTERM' do
  262. node_pids.each do |node_pid|
  263. Process.kill 'TERM', node_pid
  264. end
  265. Process.waitall
  266. raise SignalException, 'SIGTERM'
  267. end
  268. # dispatch client_ids to nodes
  269. loop do
  270. # nodes
  271. nodes_stats = Sessions::Node.stats
  272. client_ids = sessions
  273. client_ids.each do |client_id|
  274. # ask nodes for nodes
  275. next if nodes_stats[client_id]
  276. # assign to node
  277. Sessions::Node.session_assigne(client_id)
  278. sleep 1
  279. end
  280. sleep 1
  281. end
  282. end
  283. Thread.abort_on_exception = true
  284. loop do
  285. if node_id
  286. # register node
  287. Sessions::Node.register(node_id)
  288. # watch for assigned sessions
  289. client_ids = Sessions::Node.sessions_by(node_id)
  290. else
  291. client_ids = sessions
  292. end
  293. client_ids.each do |client_id|
  294. # connection already open, ignore
  295. next if @@client_threads[client_id]
  296. # get current user
  297. session_data = Sessions.get(client_id)
  298. next if session_data.blank?
  299. next if session_data[:user].blank?
  300. next if session_data[:user]['id'].blank?
  301. user = User.lookup(id: session_data[:user]['id'])
  302. next if user.blank?
  303. # start client thread
  304. next if @@client_threads[client_id].present?
  305. @@client_threads[client_id] = true
  306. @@client_threads[client_id] = Thread.new do
  307. thread_client(client_id, 0, Time.now.utc, node_id)
  308. @@client_threads[client_id] = nil
  309. log('debug', "close client (#{client_id}) thread")
  310. if ActiveRecord::Base.connection.owner == Thread.current
  311. ActiveRecord::Base.connection.close
  312. end
  313. end
  314. sleep 1
  315. end
  316. sleep 1
  317. end
  318. end
  319. =begin
  320. check if thread for client_id is running
  321. Sessions.thread_client_exists?(client_id)
  322. returns
  323. thread
  324. =end
  325. def self.thread_client_exists?(client_id)
  326. @@client_threads[client_id]
  327. end
  328. =begin
  329. start client for browser
  330. Sessions.thread_client(client_id)
  331. returns
  332. thread
  333. =end
  334. def self.thread_client(client_id, try_count = 0, try_run_time = Time.now.utc, node_id)
  335. log('debug', "LOOP #{node_id}.#{client_id} - #{try_count}")
  336. begin
  337. Sessions::Client.new(client_id, node_id)
  338. rescue => e
  339. log('error', "thread_client #{client_id} exited with error #{e.inspect}")
  340. log('error', e.backtrace.join("\n "))
  341. sleep 10
  342. begin
  343. ActiveRecord::Base.connection_pool.release_connection
  344. rescue => e
  345. log('error', "Can't reconnect to database #{e.inspect}")
  346. end
  347. try_run_max = 10
  348. try_count += 1
  349. # reset error counter if to old
  350. if try_run_time + (60 * 5) < Time.now.utc
  351. try_count = 0
  352. end
  353. try_run_time = Time.now.utc
  354. # restart job again
  355. if try_run_max > try_count
  356. thread_client(client_id, try_count, try_run_time, node_id)
  357. end
  358. raise "STOP thread_client for client #{node_id}.#{client_id} after #{try_run_max} tries"
  359. ensure
  360. ActiveSupport::CurrentAttributes.clear_all
  361. end
  362. log('debug', "/LOOP #{node_id}.#{client_id} - #{try_count}")
  363. end
  364. def self.symbolize_keys(hash)
  365. hash.each_with_object({}) do |(key, value), result|
  366. new_key = case key
  367. when String then key.to_sym
  368. else key
  369. end
  370. new_value = case value
  371. when Hash then symbolize_keys(value)
  372. else value
  373. end
  374. result[new_key] = new_value
  375. end
  376. end
  377. # we use it in rails and non rails context
  378. def self.log(level, message)
  379. if defined?(Rails)
  380. case level
  381. when 'debug'
  382. Rails.logger.debug { message }
  383. when 'info'
  384. Rails.logger.info message
  385. else
  386. Rails.logger.error message
  387. end
  388. return
  389. end
  390. puts "#{Time.now.utc.iso8601}:#{level} #{message}" # rubocop:disable Rails/Output
  391. end
  392. end