websocket_server.rb 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  1. # Copyright (C) 2012-2024 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. # By default, we are only logging errors to STDOUT.
  8. # To turn on some more logging to get some insights, please, provide one of the following parameters:
  9. # -n | --info => info
  10. # -v | --verbose => debug
  11. Rails.configuration.interface = 'websocket'
  12. AppVersion.start_maintenance_thread(process_name: 'websocket-server')
  13. EventMachine.run do
  14. EventMachine::WebSocket.start(host: @options[:b], port: @options[:p], secure: @options[:s], tls_options: @options[:tls_options]) do |ws|
  15. # register client connection
  16. ws.onopen do |handshake|
  17. WebsocketServer.onopen(ws, handshake)
  18. end
  19. # unregister client connection
  20. ws.onclose do
  21. WebsocketServer.onclose(ws)
  22. end
  23. # manage messages
  24. ws.onmessage do |msg|
  25. WebsocketServer.onmessage(ws, msg)
  26. end
  27. end
  28. # check unused connections
  29. EventMachine.add_timer(0.5) do
  30. WebsocketServer.check_unused_connections
  31. end
  32. # check open unused connections, kick all connection without activity in the last 2 minutes
  33. EventMachine.add_periodic_timer(120) do
  34. WebsocketServer.check_unused_connections
  35. end
  36. EventMachine.add_periodic_timer(20) do
  37. WebsocketServer.log_status
  38. end
  39. EventMachine.add_periodic_timer(0.4) do
  40. WebsocketServer.send_to_client
  41. end
  42. end
  43. end
  44. def self.onopen(websocket, handshake)
  45. headers = handshake.headers
  46. client_id = websocket.object_id.to_s
  47. log 'info', 'Client connected.', client_id
  48. Sessions.create(client_id, {}, { type: 'websocket' })
  49. return if @clients.include? client_id
  50. @clients[client_id] = {
  51. websocket: websocket,
  52. last_ping: Time.now.utc.to_i,
  53. error_count: 0,
  54. headers: headers,
  55. }
  56. end
  57. def self.onclose(websocket)
  58. client_id = websocket.object_id.to_s
  59. log 'info', 'Client disconnected.', client_id
  60. # removed from current client list
  61. if @clients.include? client_id
  62. @clients.delete client_id
  63. end
  64. Sessions.destroy(client_id)
  65. end
  66. def self.onmessage(websocket, msg)
  67. client_id = websocket.object_id.to_s
  68. log 'info', "receiving #{msg.to_s.bytesize} bytes", client_id
  69. log 'debug', "received: #{msg}", client_id
  70. begin
  71. log 'info', 'start: parse message to JSON', client_id
  72. data = JSON.parse(msg)
  73. log 'info', 'end: parse message to JSON', client_id
  74. rescue => e
  75. log 'error', "can't parse message: #{msg}, #{e.inspect}", client_id
  76. return
  77. end
  78. # check if connection not already exists
  79. return if !@clients[client_id]
  80. Sessions.touch(client_id) # rubocop:disable Rails/SkipsModelValidations
  81. @clients[client_id][:last_ping] = Time.now.utc.to_i
  82. if data['event']
  83. log 'info', "start: execute event '#{data['event']}'", client_id
  84. message = Sessions::Event.run(
  85. event: data['event'],
  86. payload: data,
  87. session: @clients[client_id][:session],
  88. headers: @clients[client_id][:headers],
  89. client_id: client_id,
  90. clients: @clients,
  91. options: @options,
  92. )
  93. log 'info', "end: execute event '#{data['event']}'", client_id
  94. if message
  95. websocket_send(client_id, message)
  96. end
  97. else
  98. log 'error', "unknown message '#{data.inspect}'", client_id
  99. end
  100. end
  101. def self.websocket_send(client_id, data)
  102. msg = if data.instance_of?(Array)
  103. data.to_json
  104. else
  105. "[#{data.to_json}]"
  106. end
  107. log 'info', "sending #{msg.to_s.bytesize} bytes", client_id
  108. log 'debug', "send: #{msg}", client_id
  109. if !@clients[client_id]
  110. log 'error', "no such @clients for #{client_id}", client_id
  111. return
  112. end
  113. @clients[client_id][:websocket].send(msg)
  114. end
  115. def self.check_unused_connections
  116. log 'info', 'check unused idle connections...'
  117. idle_time_in_sec = 4 * 60
  118. # close unused web socket sessions
  119. @clients.each do |client_id, client|
  120. next if (client[:last_ping].to_i + idle_time_in_sec) >= Time.now.utc.to_i
  121. log 'info', 'closing idle websocket connection', client_id
  122. # remember to not use this connection anymore
  123. client[:disconnect] = true
  124. # try to close regular
  125. client[:websocket].close_websocket
  126. # delete session from client list
  127. sleep 0.3
  128. @clients.delete(client_id)
  129. end
  130. # close unused ajax long polling sessions
  131. clients = Sessions.destroy_idle_sessions(idle_time_in_sec)
  132. clients.each do |client_id|
  133. log 'info', 'closing idle long polling connection', client_id
  134. end
  135. end
  136. def self.send_to_client
  137. return if @clients.empty?
  138. # log 'debug', 'checking for data to send...'
  139. @clients.each do |client_id, client|
  140. next if client[:disconnect]
  141. log 'debug', 'checking for data...', client_id
  142. begin
  143. queue = Sessions.queue(client_id)
  144. next if queue.blank?
  145. websocket_send(client_id, queue)
  146. rescue => e
  147. log 'error', "problem:#{e.inspect}", client_id
  148. # disconnect client
  149. client[:error_count] += 1
  150. if client[:error_count] > 20 && @clients.include?(client_id)
  151. @clients.delete client_id
  152. end
  153. end
  154. end
  155. end
  156. def self.log_status
  157. # websocket
  158. log 'info', "Status: websocket clients: #{@clients.size}"
  159. @clients.each_key do |client_id|
  160. log 'info', 'working...', client_id
  161. end
  162. # ajax
  163. client_list = Sessions.list
  164. clients = 0
  165. client_list.each_value do |client|
  166. next if client[:meta][:type] == 'websocket'
  167. clients += 1
  168. end
  169. log 'info', "Status: ajax clients: #{clients}"
  170. client_list.each do |client_id, client|
  171. next if client[:meta][:type] == 'websocket'
  172. log 'info', 'working...', client_id
  173. end
  174. end
  175. def self.log(level, data, client_id = '-')
  176. skip_log = true
  177. case level
  178. when 'error'
  179. skip_log = false
  180. when 'debug'
  181. if @options[:v]
  182. skip_log = false
  183. end
  184. when 'info'
  185. if @options[:n] || @options[:v]
  186. skip_log = false
  187. end
  188. end
  189. return if skip_log
  190. client_id_str = client_id.eql?('-') ? '' : "##{client_id}"
  191. # same format as in log/production.log
  192. puts "#{level.to_s.first.upcase}, [#{Time.now.utc.strftime('%FT%T.%6N')}#{client_id_str}] #{level.upcase} -- : #{data}" # rubocop:disable Rails/Output
  193. end
  194. end