websocket_server.rb 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239
  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. end
  100. def self.websocket_send(client_id, data)
  101. msg = if data.instance_of?(Array)
  102. data.to_json
  103. else
  104. "[#{data.to_json}]"
  105. end
  106. log 'info', "sending #{msg.to_s.bytesize} bytes", client_id
  107. log 'debug', "send: #{msg}", client_id
  108. if !@clients[client_id]
  109. log 'error', "no such @clients for #{client_id}", client_id
  110. return
  111. end
  112. @clients[client_id][:websocket].send(msg)
  113. end
  114. def self.check_unused_connections
  115. log 'info', 'check unused idle connections...'
  116. idle_time_in_sec = 4 * 60
  117. # close unused web socket sessions
  118. @clients.each do |client_id, client|
  119. next if (client[:last_ping].to_i + idle_time_in_sec) >= Time.now.utc.to_i
  120. log 'info', 'closing idle websocket connection', client_id
  121. # remember to not use this connection anymore
  122. client[:disconnect] = true
  123. # try to close regular
  124. client[:websocket].close_websocket
  125. # delete session from client list
  126. sleep 0.3
  127. @clients.delete(client_id)
  128. end
  129. # close unused ajax long polling sessions
  130. clients = Sessions.destroy_idle_sessions(idle_time_in_sec)
  131. clients.each do |client_id|
  132. log 'info', 'closing idle long polling connection', client_id
  133. end
  134. end
  135. def self.send_to_client
  136. return if @clients.empty?
  137. # log 'debug', 'checking for data to send...'
  138. @clients.each do |client_id, client|
  139. next if client[:disconnect]
  140. log 'debug', 'checking for data...', client_id
  141. begin
  142. queue = Sessions.queue(client_id)
  143. next if queue.blank?
  144. websocket_send(client_id, queue)
  145. rescue => e
  146. log 'error', "problem:#{e.inspect}", client_id
  147. # disconnect client
  148. client[:error_count] += 1
  149. if client[:error_count] > 20 && @clients.include?(client_id)
  150. @clients.delete client_id
  151. end
  152. end
  153. end
  154. end
  155. def self.log_status
  156. # websocket
  157. log 'info', "Status: websocket clients: #{@clients.size}"
  158. @clients.each_key do |client_id|
  159. log 'info', 'working...', client_id
  160. end
  161. # ajax
  162. client_list = Sessions.list
  163. clients = 0
  164. client_list.each_value do |client|
  165. next if client[:meta][:type] == 'websocket'
  166. clients += 1
  167. end
  168. log 'info', "Status: ajax clients: #{clients}"
  169. client_list.each do |client_id, client|
  170. next if client[:meta][:type] == 'websocket'
  171. log 'info', 'working...', client_id
  172. end
  173. end
  174. def self.log(level, data, client_id = '-')
  175. skip_log = true
  176. case level
  177. when 'error'
  178. skip_log = false
  179. when 'debug'
  180. if @options[:v]
  181. skip_log = false
  182. end
  183. when 'info'
  184. if @options[:n] || @options[:v]
  185. skip_log = false
  186. end
  187. end
  188. return if skip_log
  189. client_id_str = client_id.eql?('-') ? '' : "##{client_id}"
  190. # same format as in log/production.log
  191. puts "#{level.to_s.first.upcase}, [#{Time.now.utc.strftime('%FT%T.%6N')}#{client_id_str}] #{level.upcase} -- : #{data}" # rubocop:disable Rails/Output
  192. end
  193. end