<commit_before>module UState
  class Graphite
    # Forwards states to Graphite.
    HOST = '127.0.0.1'
    PORT = 2003
    INTERVAL = 5

    attr_accessor :query
    attr_accessor :host
    attr_accessor :port
    attr_accessor :interval

    def initialize(index, opts = {})
      @index = index
      @query = opts[:query]
      @host = opts[:host] || HOST
      @port = opts[:port] || PORT
      @interval = opts[:interval] || INTERVAL
      @locket = Mutex.new

      start
    end

    def connect
      @socket = TCPSocket.new(@host, @port)
    end

    def graph(q)
      if @query
        @query = "(#{@query}) or (#{q})"
      else
        @query = q
      end
    end

    def forward(state)
      string = "#{path(state)} #{state.metric} #{state.time}"
      if string["\n"]
        raise ArgumentError, "#{string} has a newline"
      end
      with_connection do |s|
        s.puts string
      end
    end

    # Formats a state into a Graphite metric path.
    def path(state)
      if state.host
        host = state.host.split('.').reverse.join('.')
        "#{host}.#{state.service.gsub(' ', '.')}"
      else
        state.service.gsub(' ', '.')
      end
    end

    def start
      @runner = Thread.new do
        loop do
          @index.query(Query.new(string: @query)).each do |state|
            forward state
          end
          sleep @interval
        end
      end
    end

    def with_connection
      tries = 0
      @locket.synchronize do
        begin
          tries += 1<commit_msg>Graphite: use adaptive time calculation.<commit_after>module UState
  class Graphite
    # Forwards states to Graphite.
    HOST = '127.0.0.1'
    PORT = 2003
    INTERVAL = 10

    attr_accessor :query
    attr_accessor :host
    attr_accessor :port
    attr_accessor :interval

    def initialize(index, opts = {})
      @index = index
      @query = opts[:query]
      @host = opts[:host] || HOST
      @port = opts[:port] || PORT
      @interval = opts[:interval] || INTERVAL
      @locket = Mutex.new

      start
    end

    def connect
      @socket = TCPSocket.new(@host, @port)
    end

    def graph(q)
      if @query
        @query = "(#{@query}) or (#{q})"
      else
        @query = q
      end
    end

    def forward(state)
      # Figure out what time to use.
      present = Time.now.to_i
      if (present - state.time) >= INTERVAL
        time = present
      else
        time = state.time
      end

      # Construct message
      string = "#{path(state)} #{state.metric} #{state.time}"
      
      # Validate string
      if string["\n"]
        raise ArgumentError, "#{string} has a newline"
      end

      with_connection do |s|
        s.puts string
      end
    end

    # Formats a state into a Graphite metric path.
    def path(state)
      if state.host
        host = state.host.split('.').reverse.join('.')
        "#{host}.#{state.service.gsub(' ', '.')}"
      else
        state.service.gsub(' ', '.')
      end
    end

    def start
      @runner = Thread.new do
        loop do
          @index.query(Query.new(string: @query)).each do |state|
            forward state
          end
          sleep @interval
        end
      end
    end

    def with_connection
      tries = 0
      @locket.synchronize do
        begin
          tries += 1
{'fraction_non_alphanumeric': 0.11647555823777912, 'fraction_numerical': 0.008147254073627036, 'mean_word_length': 1.714987714987715, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '2605484', 'n_tokens_mistral': 1056, 'n_tokens_neox': 1029, 'n_words': 285}