<commit_before>module Rcm::Cluster::Ping
  def self.ping(client, interval : Time::Span = 1.second, crt : Bool = true)
    Main.new(client, interval, crt: crt).run
  end

  class Main
    delegate nodes, to: @info

    MAX_VAL_SIZE = 256          # max terminal colum size we expect
    
    @watchers : Array(Watcher)
    @count_ch : Channel::Unbuffered(Result)
    @nodes_ch : Channel::Unbuffered(String)
    @noded_counts : Hash(NodeInfo, Array(Int64))

    def initialize(@client : Client, @interval : Time::Span, crt : Bool)
      @info = @client.cluster_info
      @count_ch  = Channel(Result).new
      @nodes_ch  = Channel(String).new
      @watchers = nodes.map{|n| Watcher.new(n, ->(){@client.new_redis(n)}, @count_ch)}
      @noded_counts = Hash(NodeInfo, Array(Int64)).new
      @show = crt ? Show::Crt.new : Show::IO.new
      @time_body = MemoryIO.new
    end

    def run
      @watchers.each(&.start(@interval))
      spawn { observe_channels }
      spawn { render }
      STDIN.gets
    end

    private def observe_channels
      receives = [@count_ch, @nodes_ch].map(&.receive_op)
      loop {
        index, value = Channel.select(receives)
        case index
        when 0
          update_count(value.as(Result))
        when 1
          update_nodes(value.as(String))
        else
          raise "[BUG] observe_channels got unexpected index(#{index})"
        end
        # sleep 0 # Comment out anyway. I don't know why this is bad
      }
    end

    private def update_nodes(nodes : String)
      @info = ClusterInfo.parse(nodes)
    end

    private def update_count(result : Result)
      counts_for(result.node) << result.count
    end

    private def render
      schedule_each(@interval) {
        shrink_counts
        shrink_time_body

        @show.clear
        @show.head(Time.now.to_s)
        @show.print("", build_time_body)
        nodes.each do |node|
          key = build_key_for(node)
          val = build_body_for(node)
          @show.print(key, val)
        end
        @show.refresh
      }
    end

    private def counts_for(node : NodeInfo)
      @noded_counts[node] ||= [] of Int64
    end

    private def build_key_for(node)
      a = counts_for(node)
      c = a.last { -1 }
      "#{node.addr}(#{c})"
    end

    private def build_body_for(node)
      ary = counts_for(node)
      last_valid = nil
      ary.map{|current|
        if current == -1
          "E"
        else
          prev = last_valid
          last_valid = current
          case prev
          when Int64
            if current > prev
              "+"
            elsif current < prev
              "-"
            else
              "."
            end
          else
            "."
          end
        end
      }.join
    end<commit_msg>ping: build_typed_key<commit_after>module Rcm::Cluster::Ping
  def self.ping(client, interval : Time::Span = 1.second, crt : Bool = true)
    Main.new(client, interval, crt: crt).run
  end

  class Main
    delegate nodes, to: @info

    MAX_VAL_SIZE = 256          # max terminal colum size we expect
    
    @watchers : Array(Watcher)
    @count_ch : Channel::Unbuffered(Result)
    @nodes_ch : Channel::Unbuffered(String)
    @noded_counts : Hash(NodeInfo, Array(Int64))

    def initialize(@client : Client, @interval : Time::Span, crt : Bool)
      @info = @client.cluster_info
      @count_ch  = Channel(Result).new
      @nodes_ch  = Channel(String).new
      @watchers = nodes.map{|n| Watcher.new(n, ->(){@client.new_redis(n)}, @count_ch)}
      @noded_counts = Hash(NodeInfo, Array(Int64)).new
      @show = crt ? Show::Crt.new : Show::IO.new
      @time_body = MemoryIO.new
    end

    def run
      @watchers.each(&.start(@interval))
      spawn { observe_channels }
      spawn { render }
      STDIN.gets
    end

    private def observe_channels
      receives = [@count_ch, @nodes_ch].map(&.receive_op)
      loop {
        index, value = Channel.select(receives)
        case index
        when 0
          update_count(value.as(Result))
        when 1
          update_nodes(value.as(String))
        else
          raise "[BUG] observe_channels got unexpected index(#{index})"
        end
        # sleep 0 # Comment out anyway. I don't know why this is bad
      }
    end

    private def update_nodes(nodes : String)
      @info = ClusterInfo.parse(nodes)
    end

    private def update_count(result : Result)
      counts_for(result.node) << result.count
    end

    private def render
      schedule_each(@interval) {
        shrink_counts
        shrink_time_body

        @show.clear
        @show.head(Time.now.to_s)
        @show.print("", build_time_body)
        @info.each_nodes do |node|
#          key = build_key_for(node)
          key = build_typed_key(node)
          val = build_body_for(node)
          @show.print(key, val)
        end
        @show.refresh
      }
    end

    private def counts_for(node : NodeInfo)
      @noded_counts[node] ||= [] of Int64
    end

    private def build_key_for(node)
      a = counts_for(node)
      c = a.last { -1 }
      "#{node.addr}(#{c})"
    end

    private def build_typed_key(node)
      c = counts_for(node).last { -1 }
      if node.serving?
        prefix = "[%-11s] " % node.slot
        "#{prefix}#{node.addr}(#{c})"
      elsif node.master?
        prefix = " ( no slots ) "
        "#{prefix}#{node.addr}(#{c})"
      else
        prefix = "    +slave    "
        "#{prefix}#{node.addr}(#{c})"
      end
    end

    private def build_body_for(node)
      ary = counts_for(node)
      last_valid = nil
      ary.map{|current|
        if current == -1
          "E"
        else
          prev = last_valid
          last_valid = current
          case prev
          when Int64
            if current > prev
              "+"
            elsif current < prev
              "-"
            else
              "."
            end
          else
            "."
          end
        end
      }.join
    end
{'fraction_non_alphanumeric': 0.11220735785953177, 'fraction_numerical': 0.006187290969899666, 'mean_word_length': 1.8576206402293358, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '4305471', 'n_tokens_mistral': 1946, 'n_tokens_neox': 1886, 'n_words': 430}