websocket_server.rb 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215
  1. # Copyright (C) 2012-2023 Zammad Foundation, https://zammad-foundation.org/
  2. class WebsocketServer
  3. cattr_reader :clients, :options
  4. def self.run(options)
  5. @options = options
  6. @clients = {}
  7. Rails.configuration.interface = 'websocket'
  8. EventMachine.run do
  9. EventMachine::WebSocket.start(host: @options[:b], port: @options[:p], secure: @options[:s], tls_options: @options[:tls_options]) do |ws|
  10. # register client connection
  11. ws.onopen do |handshake|
  12. WebsocketServer.onopen(ws, handshake)
  13. end
  14. # unregister client connection
  15. ws.onclose do
  16. WebsocketServer.onclose(ws)
  17. end
  18. # manage messages
  19. ws.onmessage do |msg|
  20. WebsocketServer.onmessage(ws, msg)
  21. end
  22. end
  23. # check unused connections
  24. EventMachine.add_timer(0.5) do
  25. WebsocketServer.check_unused_connections
  26. end
  27. # check open unused connections, kick all connection without activity in the last 2 minutes
  28. EventMachine.add_periodic_timer(120) do
  29. WebsocketServer.check_unused_connections
  30. end
  31. EventMachine.add_periodic_timer(20) do
  32. WebsocketServer.log_status
  33. end
  34. EventMachine.add_periodic_timer(0.4) do
  35. WebsocketServer.send_to_client
  36. end
  37. end
  38. end
  39. def self.onopen(websocket, handshake)
  40. headers = handshake.headers
  41. client_id = websocket.object_id.to_s
  42. log 'info', 'Client connected.', client_id
  43. Sessions.create(client_id, {}, { type: 'websocket' })
  44. return if @clients.include? client_id
  45. @clients[client_id] = {
  46. websocket: websocket,
  47. last_ping: Time.now.utc.to_i,
  48. error_count: 0,
  49. headers: headers,
  50. }
  51. end
  52. def self.onclose(websocket)
  53. client_id = websocket.object_id.to_s
  54. log 'info', 'Client disconnected.', client_id
  55. # removed from current client list
  56. if @clients.include? client_id
  57. @clients.delete client_id
  58. end
  59. Sessions.destroy(client_id)
  60. end
  61. def self.onmessage(websocket, msg)
  62. client_id = websocket.object_id.to_s
  63. log 'debug', "received: #{msg} ", client_id
  64. begin
  65. data = JSON.parse(msg)
  66. rescue => e
  67. log 'error', "can't parse message: #{msg}, #{e.inspect}", client_id
  68. return
  69. end
  70. # check if connection not already exists
  71. return if !@clients[client_id]
  72. Sessions.touch(client_id) # rubocop:disable Rails/SkipsModelValidations
  73. @clients[client_id][:last_ping] = Time.now.utc.to_i
  74. if data['event']
  75. log 'debug', "execute event '#{data['event']}'", client_id
  76. message = Sessions::Event.run(
  77. event: data['event'],
  78. payload: data,
  79. session: @clients[client_id][:session],
  80. headers: @clients[client_id][:headers],
  81. client_id: client_id,
  82. clients: @clients,
  83. options: @options,
  84. )
  85. if message
  86. websocket_send(client_id, message)
  87. end
  88. else
  89. log 'error', "unknown message '#{data.inspect}'", client_id
  90. end
  91. ensure
  92. ActiveSupport::CurrentAttributes.clear_all
  93. end
  94. def self.websocket_send(client_id, data)
  95. msg = if data.instance_of?(Array)
  96. data.to_json
  97. else
  98. "[#{data.to_json}]"
  99. end
  100. log 'debug', "send #{msg}", client_id
  101. if !@clients[client_id]
  102. log 'error', "no such @clients for #{client_id}", client_id
  103. return
  104. end
  105. @clients[client_id][:websocket].send(msg)
  106. end
  107. def self.check_unused_connections
  108. log 'info', 'check unused idle connections...'
  109. idle_time_in_sec = 4 * 60
  110. # close unused web socket sessions
  111. @clients.each do |client_id, client|
  112. next if (client[:last_ping].to_i + idle_time_in_sec) >= Time.now.utc.to_i
  113. log 'info', 'closing idle websocket connection', client_id
  114. # remember to not use this connection anymore
  115. client[:disconnect] = true
  116. # try to close regular
  117. client[:websocket].close_websocket
  118. # delete session from client list
  119. sleep 0.3
  120. @clients.delete(client_id)
  121. end
  122. # close unused ajax long polling sessions
  123. clients = Sessions.destroy_idle_sessions(idle_time_in_sec)
  124. clients.each do |client_id|
  125. log 'info', 'closing idle long polling connection', client_id
  126. end
  127. end
  128. def self.send_to_client
  129. return if @clients.empty?
  130. # log 'debug', 'checking for data to send...'
  131. @clients.each do |client_id, client|
  132. next if client[:disconnect]
  133. log 'debug', 'checking for data...', client_id
  134. begin
  135. queue = Sessions.queue(client_id)
  136. next if queue.blank?
  137. log 'info', 'send data to client', client_id
  138. websocket_send(client_id, queue)
  139. rescue => e
  140. log 'error', "problem:#{e.inspect}", client_id
  141. # disconnect client
  142. client[:error_count] += 1
  143. if client[:error_count] > 20 && @clients.include?(client_id)
  144. @clients.delete client_id
  145. end
  146. end
  147. end
  148. end
  149. def self.log_status
  150. # websocket
  151. log 'info', "Status: websocket clients: #{@clients.size}"
  152. @clients.each_key do |client_id|
  153. log 'info', 'working...', client_id
  154. end
  155. # ajax
  156. client_list = Sessions.list
  157. clients = 0
  158. client_list.each_value do |client|
  159. next if client[:meta][:type] == 'websocket'
  160. clients += 1
  161. end
  162. log 'info', "Status: ajax clients: #{clients}"
  163. client_list.each do |client_id, client|
  164. next if client[:meta][:type] == 'websocket'
  165. log 'info', 'working...', client_id
  166. end
  167. end
  168. def self.log(level, data, client_id = '-')
  169. return if !@options[:v] && level == 'debug'
  170. puts "#{Time.now.utc.iso8601}:client(#{client_id}) #{data}" # rubocop:disable Rails/Output
  171. # puts "#{Time.now.utc.iso8601}:#{ level }:client(#{ client_id }) #{ data }"
  172. end
  173. end