<commit_before>require 'eventmachine'
require 'faye/websocket'

module JanusGateway

  class Transport::WebSocket < Transport

    attr_reader :transaction_queue

    # @param [String] url
    # @param [String] protocol
    def initialize(url, protocol = 'janus-protocol')
      @url = url
      @protocol = protocol
      @client = nil
      @transaction_queue = Hash.new
    end

    def connect
      EventMachine.run do

        EM.error_handler { |e| raise(e) }

        @client = _create_client(@url, @protocol)

        @client.on :open do
          self.emit :open
        end

        @client.on :message do |event|
          data = JSON.parse(event.data)

          transaction_list = @transaction_queue.clone

          transaction_id = data['transaction']
          unless transaction_id.nil?
            promise = transaction_list[transaction_id]
            unless promise.nil?
              if ['success', 'ack'].include?(data['janus'])
                promise.set(data)
                promise.execute
              else
                error_data = data['error']
                error = JanusGateway::Error.new(error_data['code'], error_data['reason'])
                promise.fail(error).execute
              end
            end
          end

          self.emit :message, data
        end

        @client.on :close do
          self.emit :close
        end
      end
    end

    # @param [Hash] data
    def send(data)
      @client.send(JSON.generate(data))
    end

    # @param [Hash] data
    # @return [Concurrent::Promise]
    def send_transaction(data)
      promise = Concurrent::Promise.new
      transaction = transaction_id_new

      data[:transaction] = transaction
      send(data)

      @transaction_queue[transaction] = promise

      Thread.new do
        sleep(_transaction_timeout)
        error = JanusGateway::Error.new(0, "Transaction id `#{transaction}` has failed due to timeout!")
        promise.fail(error).execute
        @transaction_queue.remove(transaction)
      end

      promise
    end

    def disconnect
      @client.close
      EventMachine.stop
    end

    # @return [Faye::WebSocket::API::CONNECTING, Faye::WebSocket::API::OPEN, Faye::WebSocket::API::CLOSING, Faye::WebSocket::API::CLOSED]
    def ready_state
      @client.ready_state unless @client.nil?
    end

    # @return [TrueClass, FalseClass]
    def is_connected?
      (@client.nil? == false) and (@client.ready_state == Faye::WebSocket::API::OPEN)<commit_msg>Fixed @client usages in websocket transport<commit_after>require 'eventmachine'
require 'faye/websocket'

module JanusGateway

  class Transport::WebSocket < Transport

    attr_reader :transaction_queue

    # @param [String] url
    # @param [String] protocol
    def initialize(url, protocol = 'janus-protocol')
      @url = url
      @protocol = protocol
      @client = nil
      @transaction_queue = Hash.new
    end

    def connect
      EventMachine.run do

        EM.error_handler { |e| raise(e) }

        @client = _create_client(@url, @protocol)

        @client.on :open do
          self.emit :open
        end

        @client.on :message do |event|
          data = JSON.parse(event.data)

          transaction_list = @transaction_queue.clone

          transaction_id = data['transaction']
          unless transaction_id.nil?
            promise = transaction_list[transaction_id]
            unless promise.nil?
              if ['success', 'ack'].include?(data['janus'])
                promise.set(data)
                promise.execute
              else
                error_data = data['error']
                error = JanusGateway::Error.new(error_data['code'], error_data['reason'])
                promise.fail(error).execute
              end
            end
          end

          self.emit :message, data
        end

        @client.on :close do
          self.emit :close
        end
      end
    end

    # @param [Hash] data
    def send(data)
      client.send(JSON.generate(data))
    end

    # @param [Hash] data
    # @return [Concurrent::Promise]
    def send_transaction(data)
      promise = Concurrent::Promise.new
      transaction = transaction_id_new

      data[:transaction] = transaction
      send(data)

      @transaction_queue[transaction] = promise

      Thread.new do
        sleep(_transaction_timeout)
        error = JanusGateway::Error.new(0, "Transaction id `#{transaction}` has failed due to timeout!")
        promise.fail(error).execute
        @transaction_queue.remove(transaction)
      end

      promise
    end

    def disconnect
      client.close
      EventMachine.stop
    end

    # @return [Faye::WebSocket::API::CONNECTING, Faye::WebSocket::API::OPEN, Faye::WebSocket::API::CLOSING, Faye::WebSocket::API::CLOSED]
    def ready_state
      client.ready_state unless client.nil?
    end

    # @return [TrueClass, FalseClass]
    def is_connected?
      (client.nil? == false) and (ready_state == Faye::WebSocket::API::OPEN)
{'fraction_non_alphanumeric': 0.1037037037037037, 'fraction_numerical': 0.0004004004004004004, 'mean_word_length': 2.208734746307001, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '398976', 'n_tokens_mistral': 1419, 'n_tokens_neox': 1393, 'n_words': 316}