websocket_server.rb 6.5 KB

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