zammad/lib/websocket_server.rb

# Copyright (C) 2012-2022 Zammad Foundation, https://zammad-foundation.org/

class WebsocketServer

  cattr_reader :clients, :options

  def self.run(options)
    @options = options
    @clients = {}

    Rails.configuration.interface = 'websocket'
    EventMachine.run do
      EventMachine::WebSocket.start(host: @options[:b], port: @options[:p], secure: @options[:s], tls_options: @options[:tls_options]) do |ws|

        # register client connection
        ws.onopen do |handshake|
          WebsocketServer.onopen(ws, handshake)
        end

        # unregister client connection
        ws.onclose do
          WebsocketServer.onclose(ws)
        end

        # manage messages
        ws.onmessage do |msg|
          WebsocketServer.onmessage(ws, msg)
        end
      end

      # check unused connections
      EventMachine.add_timer(0.5) do
        WebsocketServer.check_unused_connections
      end

      # check open unused connections, kick all connection without activity in the last 2 minutes
      EventMachine.add_periodic_timer(120) do
        WebsocketServer.check_unused_connections
      end

      EventMachine.add_periodic_timer(20) do
        WebsocketServer.log_status
      end

      EventMachine.add_periodic_timer(0.4) do
        WebsocketServer.send_to_client
      end
    end
  end

  def self.onopen(websocket, handshake)
    headers = handshake.headers
    client_id = websocket.object_id.to_s
    log 'info', 'Client connected.', client_id
    Sessions.create(client_id, {}, { type: 'websocket' })

    return if @clients.include? client_id

    @clients[client_id] = {
      websocket:   websocket,
      last_ping:   Time.now.utc.to_i,
      error_count: 0,
      headers:     headers,
    }
  end

  def self.onclose(websocket)
    client_id = websocket.object_id.to_s
    log 'info', 'Client disconnected.', client_id

    # removed from current client list
    if @clients.include? client_id
      @clients.delete client_id
    end

    Sessions.destroy(client_id)
  end

  def self.onmessage(websocket, msg)
    client_id = websocket.object_id.to_s
    log 'debug', "received: #{msg} ", client_id
    begin
      data = JSON.parse(msg)
    rescue => e
      log 'error', "can't parse message: #{msg}, #{e.inspect}", client_id
      return
    end

    # check if connection not already exists
    return if !@clients[client_id]

    Sessions.touch(client_id) # rubocop:disable Rails/SkipsModelValidations
    @clients[client_id][:last_ping] = Time.now.utc.to_i

    # spool messages for new connects
    if data['spool']
      Sessions.spool_create(data)
    end

    if data['event']
      log 'debug', "execute event '#{data['event']}'", client_id
      message = Sessions::Event.run(
        event:     data['event'],
        payload:   data,
        session:   @clients[client_id][:session],
        headers:   @clients[client_id][:headers],
        client_id: client_id,
        clients:   @clients,
        options:   @options,
      )
      if message
        websocket_send(client_id, message)
      end
    else
      log 'error', "unknown message '#{data.inspect}'", client_id
    end
  ensure
    ActiveSupport::CurrentAttributes.clear_all
  end

  def self.websocket_send(client_id, data)
    msg = if data.instance_of?(Array)
            data.to_json
          else
            "[#{data.to_json}]"
          end
    log 'debug', "send #{msg}", client_id
    if !@clients[client_id]
      log 'error', "no such @clients for #{client_id}", client_id
      return
    end
    @clients[client_id][:websocket].send(msg)
  end

  def self.check_unused_connections
    log 'info', 'check unused idle connections...'

    idle_time_in_sec = 4 * 60

    # close unused web socket sessions
    @clients.each do |client_id, client|

      next if (client[:last_ping].to_i + idle_time_in_sec) >= Time.now.utc.to_i

      log 'info', 'closing idle websocket connection', client_id

      # remember to not use this connection anymore
      client[:disconnect] = true

      # try to close regular
      client[:websocket].close_websocket

      # delete session from client list
      sleep 0.3
      @clients.delete(client_id)
    end

    # close unused ajax long polling sessions
    clients = Sessions.destroy_idle_sessions(idle_time_in_sec)
    clients.each do |client_id|
      log 'info', 'closing idle long polling connection', client_id
    end
  end

  def self.send_to_client
    return if @clients.size.zero?

    # log 'debug', 'checking for data to send...'
    @clients.each do |client_id, client|
      next if client[:disconnect]

      log 'debug', 'checking for data...', client_id
      begin
        queue = Sessions.queue(client_id)
        next if queue.blank?

        log 'info', 'send data to client', client_id
        websocket_send(client_id, queue)
      rescue => e
        log 'error', "problem:#{e.inspect}", client_id

        # disconnect client
        client[:error_count] += 1
        if client[:error_count] > 20 && @clients.include?(client_id)
          @clients.delete client_id
        end
      end
    end
  end

  def self.log_status
    # websocket
    log 'info', "Status: websocket clients: #{@clients.size}"
    @clients.each_key do |client_id|
      log 'info', 'working...', client_id
    end

    # ajax
    client_list = Sessions.list
    clients = 0
    client_list.each_value do |client|
      next if client[:meta][:type] == 'websocket'

      clients += 1
    end
    log 'info', "Status: ajax clients: #{clients}"
    client_list.each do |client_id, client|
      next if client[:meta][:type] == 'websocket'

      log 'info', 'working...', client_id
    end
  end

  def self.log(level, data, client_id = '-')
    return if !@options[:v] && level == 'debug'

    puts "#{Time.now.utc.iso8601}:client(#{client_id}) #{data}" # rubocop:disable Rails/Output
    # puts "#{Time.now.utc.iso8601}:#{ level }:client(#{ client_id }) #{ data }"
  end
end