<commit_before>require 'faye/websocket'
require 'eventmachine'
require 'base64'
require_relative '../logging'
require_relative './rpc_server'

module Cloud
  class WebsocketClient
    class Config
      attr_accessor :api_uri
      def initialize
        @api_uri = nil
      end
    end

    def self.configure(&block)
      config = Config.new
      yield config
      @@api_uri = config.api_uri
    end

    def self.api_uri
      @@api_uri
    end

    include Logging
    KEEPALIVE_TIME = 30
    @@api_uri
    attr_reader :api_uri,
                :client_id,
                :client_secret,
                :ws,
                :rpc_server,
                :ping_timer

    delegate :on, to: :ws


    ##
    # @param [String] api_uri
    # @param [String] client_id
    # @param [String] client_secret
    def initialize(client_id, client_secret)
      @api_uri = self.class.api_uri
      @client_id = client_id
      @client_secret = client_secret
      @rpc_server = RpcServer.new
      @connected = false
      @connecting = false
      @ping_timer = nil
    end

    def ensure_connect
      @connect_timer = EM::PeriodicTimer.new(5) {
        connect unless connected?
      }
      @connect_verify_timer = EM::PeriodicTimer.new(KEEPALIVE_TIME) {
        if connected?
          EM.next_tick { verify_connection }
        end
      }
    end

    def disconnect
      @connect_timer.cancel
      @connect_verify_timer.cancel
      self.ws.close if self.ws
    end

    # @return [Boolean]
    def connected?
      @connected
    end

    # @return [Boolean]
    def connecting?
      @connecting
    end

    def connect
      return if connecting?
      if self.api_uri.to_s.empty?
        error "Cloud Socket URI not configured"
        return
      end
      @connected = false
      @connecting = true
      headers = {
        'Authorization' => "Basic #{Base64.urlsafe_encode64(self.client_id+':'+self.client_secret)}"
      }
      @ws = Faye::WebSocket::Client.new("#{self.api_uri}/platform", nil, { headers: headers })

      @ws.on :open do |event|
        on_open(event)
      end
      @ws.on :message do |event|
        on_message(event)
      end
      @ws.on :close do |event|
        on_close(event)
      end
      @ws.on :error do |event|
        error "cloud connection closed with error: #{event.message}"
      end

    end

    ##
    # @param [String, Array] msg
    def send_message(msg)
      EM.next_tick {
        begin
          @ws.send(msg) if @ws
        rescue
          error "failed to send message"
        end
      }
    rescue => exc
      error "failed to send message: #{exc.message}"
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_open(event)
      ping_timer.cancel if ping_timer
      info "cloud connection opened to #{self.api_uri}"
      @connected = true
      @connecting = false
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_message(event)
      data = MessagePack.unpack(event.data.pack('c*'))
      if request_message?(data)
        EM.defer {
          response = rpc_server.handle_request(data)
          send_message(MessagePack.dump(response).bytes)
        }
      elsif notification_message?(data)
        EM.defer {
          rpc_server.handle_notification(data)
        }
      end
    rescue => exc
      error exc.message
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_close(event)
      @connected = false
      @connecting = false
      @ws = nil
      if event.code == 1002
        handle_invalid_token
      end
      info "cloud connection closed with code: #{event.code}"
    rescue => exc
      error exc.message<commit_msg>Execute cloud rpc requests only on leader instance (#1399)<commit_after>require 'faye/websocket'
require 'eventmachine'
require 'base64'
require_relative '../logging'
require_relative './rpc_server'
require_relative '../../helpers/current_leader'
module Cloud
  class WebsocketClient
    include CurrentLeader
    class Config
      attr_accessor :api_uri
      def initialize
        @api_uri = nil
      end
    end

    def self.configure(&block)
      config = Config.new
      yield config
      @@api_uri = config.api_uri
    end

    def self.api_uri
      @@api_uri
    end

    include Logging
    KEEPALIVE_TIME = 30
    @@api_uri
    attr_reader :api_uri,
                :client_id,
                :client_secret,
                :ws,
                :rpc_server,
                :ping_timer

    delegate :on, to: :ws


    ##
    # @param [String] api_uri
    # @param [String] client_id
    # @param [String] client_secret
    def initialize(client_id, client_secret)
      @api_uri = self.class.api_uri
      @client_id = client_id
      @client_secret = client_secret
      @rpc_server = RpcServer.new
      @connected = false
      @connecting = false
      @ping_timer = nil
    end

    def ensure_connect
      @connect_timer = EM::PeriodicTimer.new(5) {
        connect unless connected?
      }
      @connect_verify_timer = EM::PeriodicTimer.new(KEEPALIVE_TIME) {
        if connected?
          EM.next_tick { verify_connection }
        end
      }
    end

    def disconnect
      @connect_timer.cancel
      @connect_verify_timer.cancel
      self.ws.close if self.ws
    end

    # @return [Boolean]
    def connected?
      @connected
    end

    # @return [Boolean]
    def connecting?
      @connecting
    end

    def connect
      return if connecting?
      if self.api_uri.to_s.empty?
        error "Cloud Socket URI not configured"
        return
      end
      @connected = false
      @connecting = true
      headers = {
        'Authorization' => "Basic #{Base64.urlsafe_encode64(self.client_id+':'+self.client_secret)}"
      }
      @ws = Faye::WebSocket::Client.new("#{self.api_uri}/platform", nil, { headers: headers })

      @ws.on :open do |event|
        on_open(event)
      end
      @ws.on :message do |event|
        on_message(event)
      end
      @ws.on :close do |event|
        on_close(event)
      end
      @ws.on :error do |event|
        error "cloud connection closed with error: #{event.message}"
      end

    end

    ##
    # @param [String, Array] msg
    def send_message(msg)
      EM.next_tick {
        begin
          @ws.send(msg) if @ws
        rescue
          error "failed to send message"
        end
      }
    rescue => exc
      error "failed to send message: #{exc.message}"
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_open(event)
      ping_timer.cancel if ping_timer
      info "cloud connection opened to #{self.api_uri}"
      @connected = true
      @connecting = false
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_message(event)
        data = MessagePack.unpack(event.data.pack('c*'))
        if request_message?(data)
          EM.defer {
            if leader?
              response = rpc_server.handle_request(data)
              send_message(MessagePack.dump(response).bytes)
            end
          }
        elsif notification_message?(data)
          EM.defer {
            if leader?
              rpc_server.handle_notification(data)
            end
          }
        end
    rescue => exc
      error exc.message
    end

    # @param [Faye::WebSocket::API::Event] event
    def on_close(event)
      @connected = false
      @connecting = false
      @ws = nil
      if event.code == 1002
        handle_invalid_token
      end
      info "cloud connection closed with code: #{event.code}"
    rescue => exc
      error exc.message
{'fraction_non_alphanumeric': 0.09410520750726936, 'fraction_numerical': 0.003965107057890563, 'mean_word_length': 1.9873667587840504, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '519545', 'n_tokens_mistral': 2339, 'n_tokens_neox': 2273, 'n_words': 557}