<commit_before>module RethinkDB
  module Server
    class Client
      property streams = {} of UInt64 => RethinkDB::Cursor

      def initialize(@conn : RethinkDB::Connection)
      end

      def close
        @streams.each_value &.close
      end

      def execute(query_id : UInt64, message : Array)
        start = Time.utc

        begin
          answer = nil
          case message[0]
          when 1 # START
            term = ReQL::Term.parse(message[1])
            runopts = message[2]?.try &.as_h? || {} of String => JSON::Any

            result = @conn.run(term, RunOpts.new(runopts))
            case result
            when RethinkDB::Cursor
              list = result.first(40).to_a
              if list.size == 40
                @streams[query_id] = result
              end
              answer = {
                "t" => list.size == 40 ? 3 : 2,
                "r" => list,
                "n" => [] of String,
                "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
              }
            when RethinkDB::Datum
              answer = {
                "t" => 1,
                "r" => [result],
                "n" => [] of String,
                "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
              }
            else
              raise "BUG"
            end
          when 2 # CONTINUE
            result = @streams[query_id]
            list = result.first(40).to_a
            if list.size < 40
              @streams.delete query_id
            end
            answer = {
              "t" => list.size == 40 ? 3 : 2,
              "r" => list,
              "n" => [] of String,
              "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
            }
          when 3 # STOP
            result = @streams[query_id]?
            if result
              result.close
              @streams.delete query_id
            end
            answer = {
              "t" => 2,
              "r" => [] of String,
              "n" => [] of String,
            }
          when 4 # NOREPLY_WAIT
          when 5 # SERVER_INFO
            info = {
              "id"    => "aa", # Storage::Config.server_info.name,
              "name"  => "aa", # Storage::Config.server_info.name,
              "proxy" => false,
            }
            answer = {"t" => 5, "r" => [info]}
          end

          if !answer
            raise ReQL::InternalError.new "Invalid type of query."
          end

          return answer.to_json
        rescue ex : ReQL::CompileError
          return {"t" => 17, "r" => [ex.message], "b" => [] of String}.to_json
        rescue ex : ReQL::RuntimeError
          error_type = case ex
                       when ReQL::InternalError       ; 1000000
                       when ReQL::ResourceLimitError  ; 2000000
                       when ReQL::QueryLogicError     ; 3000000
                       when ReQL::NonExistenceError   ; 3100000
                       when ReQL::OpFailedError       ; 4100000
                       when ReQL::OpIndeterminateError; 4200000
                       when ReQL::UserError           ; 5000000
                       when ReQL::PermissionError     ; 6000000
                       else                             0
                       end
          return {"t" => 18, "e" => error_type, "r" => [ex.message], "b" => [] of String}.to_json
        rescue ex
          ex.inspect_with_backtrace
          return {"t" => 18, "e" => 1000000, "r" => [ex.message], "b" => [] of String}.to_json
        end

        return "BUG"
      end
    end
  end
end<commit_msg>stub NOREPLY_WAIT implementation<commit_after>module RethinkDB
  module Server
    class Client
      property streams = {} of UInt64 => RethinkDB::Cursor

      def initialize(@conn : RethinkDB::Connection)
      end

      def close
        @streams.each_value &.close
      end

      def execute(query_id : UInt64, message : Array)
        start = Time.utc

        begin
          answer = nil
          case message[0]
          when 1 # START
            term = ReQL::Term.parse(message[1])
            runopts = message[2]?.try &.as_h? || {} of String => JSON::Any

            result = @conn.run(term, RunOpts.new(runopts))
            case result
            when RethinkDB::Cursor
              list = result.first(40).to_a
              if list.size == 40
                @streams[query_id] = result
              end
              answer = {
                "t" => list.size == 40 ? 3 : 2,
                "r" => list,
                "n" => [] of String,
                "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
              }
            when RethinkDB::Datum
              answer = {
                "t" => 1,
                "r" => [result],
                "n" => [] of String,
                "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
              }
            else
              raise "BUG"
            end
          when 2 # CONTINUE
            result = @streams[query_id]
            list = result.first(40).to_a
            if list.size < 40
              @streams.delete query_id
            end
            answer = {
              "t" => list.size == 40 ? 3 : 2,
              "r" => list,
              "n" => [] of String,
              "p" => [{"duration(ms)" => (Time.utc - start).to_f * 1000}],
            }
          when 3 # STOP
            result = @streams[query_id]?
            if result
              result.close
              @streams.delete query_id
            end
            answer = {
              "t" => 2,
              "r" => [] of String,
              "n" => [] of String,
            }
          when 4 # NOREPLY_WAIT
            # TODO: Wait for previous noreply operations to finish
            answer = {
              "t" => 4,
              "r" => [] of String,
            }
          when 5 # SERVER_INFO
            info = {
              "id"    => "aa", # Storage::Config.server_info.name,
              "name"  => "aa", # Storage::Config.server_info.name,
              "proxy" => false,
            }
            answer = {"t" => 5, "r" => [info]}
          else
            raise ReQL::InternalError.new "Invalid type of query: #{message.inspect}"
          end

          return answer.to_json
        rescue ex : ReQL::CompileError
          return {"t" => 17, "r" => [ex.message], "b" => [] of String}.to_json
        rescue ex : ReQL::RuntimeError
          error_type = case ex
                       when ReQL::InternalError       ; 1000000
                       when ReQL::ResourceLimitError  ; 2000000
                       when ReQL::QueryLogicError     ; 3000000
                       when ReQL::NonExistenceError   ; 3100000
                       when ReQL::OpFailedError       ; 4100000
                       when ReQL::OpIndeterminateError; 4200000
                       when ReQL::UserError           ; 5000000
                       when ReQL::PermissionError     ; 6000000
                       else                             0
                       end
          return {"t" => 18, "e" => error_type, "r" => [ex.message], "b" => [] of String}.to_json
        rescue ex
          ex.inspect_with_backtrace
          return {"t" => 18, "e" => 1000000, "r" => [ex.message], "b" => [] of String}.to_json
        end
      end
    end
  end
end
{'fraction_non_alphanumeric': 0.11874492832025967, 'fraction_numerical': 0.030700568028130915, 'mean_word_length': 1.1440997390547984, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3266161', 'n_tokens_mistral': 2279, 'n_tokens_neox': 2035, 'n_words': 508}