class Selenium::WebDriver::WebSocketConnection

# File lib/selenium/webdriver/common/websocket_connection.rb, line 38
def initialize(url:)
  @callback_threads = ThreadGroup.new

  @callbacks_mtx = Mutex.new
  @messages_mtx = Mutex.new
  @closing_mtx = Mutex.new

  @closing = false
  @session_id = nil
  @url = url

  process_handshake
  @socket_thread = attach_socket_listener
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 80
def add_callback(event, &block)
  @callbacks_mtx.synchronize do
    callbacks[event] << block
    block.object_id
  end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 76
def callbacks
  @callbacks ||= Hash.new { |callbacks, event| callbacks[event] = [] }
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 53
def close
  @closing_mtx.synchronize do
    return if @closing

    @closing = true
  end

  begin
    socket.close
  rescue *CONNECTION_ERRORS => e
    WebDriver.logger.debug "WebSocket listener closed: #{e.class}: #{e.message}", id: :ws
    # already closed
  end

  # Let threads unwind instead of calling exit
  @socket_thread&.join(0.5)
  @callback_threads.list.each do |thread|
    thread.join(0.5)
  rescue StandardError => e
    WebDriver.logger.debug "Failed to join thread during close: #{e.class}: #{e.message}", id: :ws
  end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 87
def remove_callback(event, id)
  @callbacks_mtx.synchronize do
    return if @closing

    callbacks_for_event = callbacks[event]
    return if callbacks_for_event.reject! { |cb| cb.object_id == id }

    ids = callbacks_for_event.map(&:object_id)
    raise Error::WebDriverError, "Callback with ID #{id} does not exist for event #{event}: #{ids}"
  end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 99
def send_cmd(**payload)
  id = next_id
  data = payload.merge(id: id)
  WebDriver.logger.debug "WebSocket -> #{data}"[...MAX_LOG_MESSAGE_SIZE], id: :ws
  data = JSON.generate(data)
  out_frame = WebSocket::Frame::Outgoing::Client.new(version: ws.version, data: data, type: 'text')

  begin
    socket.write(out_frame.to_s)
  rescue *CONNECTION_ERRORS => e
    raise e, "WebSocket is closed (#{e.class}: #{e.message})"
  end

  wait.until { @messages_mtx.synchronize { messages.delete(id) } }
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 126
def attach_socket_listener
  Thread.new do
    Thread.current.report_on_exception = false

    loop do
      break if @closing

      incoming_frame << socket.readpartial(1024)

      while (frame = incoming_frame.next)
        break if @closing

        message = process_frame(frame)
        next unless message['method']

        @messages_mtx.synchronize { callbacks[message['method']].dup }.each do |callback|
          @callback_threads.add(callback_thread(message['params'], &callback))
        end
      end
    end
  rescue *CONNECTION_ERRORS, WebSocket::Error => e
    WebDriver.logger.debug "WebSocket listener closed: #{e.class}: #{e.message}", id: :ws
  end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 168
def callback_thread(params)
  Thread.new do
    Thread.current.abort_on_exception = false
    Thread.current.report_on_exception = false
    next if @closing

    yield params
  rescue Error::WebDriverError, *CONNECTION_ERRORS => e
    WebDriver.logger.debug "Callback aborted: #{e.class}: #{e.message}", id: :ws
  rescue StandardError => e
    next if @closing

    bt = Array(e.backtrace).first(5).join("\n")
    WebDriver.logger.error "Callback error: #{e.class}: #{e.message}\n#{bt}", id: :ws
  end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 151
def incoming_frame
  @incoming_frame ||= WebSocket::Frame::Incoming::Client.new(version: ws.version)
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 117
def messages
  @messages ||= {}
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 206
def next_id
  @id ||= 0
  @id += 1
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 155
def process_frame(frame)
  message = frame.to_s

  # Firefox will periodically fail on unparsable empty frame
  return {} if message.empty?

  msg = JSON.parse(message)
  @messages_mtx.synchronize { messages[msg['id']] = msg if msg.key?('id') }

  WebDriver.logger.debug "WebSocket <- #{msg}"[...MAX_LOG_MESSAGE_SIZE], id: :ws
  msg
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 121
def process_handshake
  socket.print(ws.to_s)
  ws << socket.readpartial(1024) until ws.finished?
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 189
def socket
  @socket ||= if URI(@url).scheme == 'wss'
                socket = TCPSocket.new(ws.host, ws.port)
                socket = OpenSSL::SSL::SSLSocket.new(socket, OpenSSL::SSL::SSLContext.new)
                socket.sync_close = true
                socket.connect

                socket
              else
                TCPSocket.new(ws.host, ws.port)
              end
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 185
def wait
  @wait ||= Wait.new(timeout: RESPONSE_WAIT_TIMEOUT, interval: RESPONSE_WAIT_INTERVAL)
end

# File lib/selenium/webdriver/common/websocket_connection.rb, line 202
def ws
  @ws ||= WebSocket::Handshake::Client.new(url: @url)
end