Update NSQ code (#16)

* Working on a JSON query lang

The idea being that you can make a call into toolkit with a query like

  select .foo from json where .id = "123"

So you can easily filter JSON files. This is like JQ, but not intended
to be any reformat, or complex filtering. This is supposed to be a
simple filter tool.

* Refactor all of NSQ

* Update musl-builder.Dockerfile

* API calls are async, and don't require nightly

Author: ethankhall

Cargo.lock

@@ -27,4 +27,8 @@ kopy-common-lib = { git = "https://github.com/ethankhall/rust-shared.git", rev =
 prettytable-rs = "0.8"
 termion = "*"
 lazy_static = "*"
-flate2 = "1.0"
+flate2 = "1.0"
+atty = "0.2"
+pest = "2.1.0"
+pest_derive = "2.1.0"
+tokio = "0.1"

Cargo.toml

@@ -6,5 +6,6 @@ RUN chmod +x ~/bin/crom
 ADD . ./
 RUN sudo chown -R rust:rust .
 RUN ~/bin/crom update-version --pre-release release
+RUN cargo --version
 
 CMD cargo test && cargo build --release

ci/musl-builder.Dockerfile

@@ -89,10 +89,14 @@ subcommands:
       settings:
         - SubcommandRequiredElseHelp
       subcommands:
-        - status:
-            about: Shows the status of NSQ Topic
-            settings:
-              - ArgRequiredElseHelp
+        - stats:
+            about: Shows the stats on NSQ
+            groups: 
+                - filter:
+                    required: true
+                    args:
+                        - producers
+                        - topics
             args:
                 - count:
                     help: How many times should I report back?
@@ -104,19 +108,26 @@ subcommands:
                     long: lookupd-host
                     takes_value: true
                     required: true
+                    env: NSQ_LOOKUP_HOST
                 - nsq_lookup_port:
                     help: Port to NSQ Lookup
                     long: lookupd-port
                     takes_value: true
                     default_value: "4161"
+                    env: NSQ_LOOKUP_PORT
                 - delay:
                     help: Ideally, how long between checks to NSQ, in seconds.
                     long: delay
                     takes_value: true
                     default_value: "1"
-                - TOPIC:
-                    help: Which topic should be check on
-                    required: true
+                - producers:
+                    help: Limit to specific producers
+                    long: filter-producers
+                    takes_value: true
+                    multiple: true
+                - topics:
+                    help: Limit to specific topics
+                    long: filter-topics
                     takes_value: true
                     multiple: true
         - send:
@@ -165,6 +176,22 @@ subcommands:
       settings:
             - SubcommandRequiredElseHelp
       subcommands:
+        - sql:
+            about: Filter new-line delemited JSON using a SQL like expression. 
+            settings:
+                - ArgRequiredElseHelp
+                - Hidden
+            args:
+                - EXP:
+                    help: SQL expression
+                    required: true
+                    takes_value: true
+                    last: true
+                - json:
+                    help: Path to file to process
+                    takes_value: true
+                    required: true
+                    long: json
         - latest:
             about: Filter new-line delemited JSON stream to the newest message
             long_about: If a JSON blob has both an ID that's unique, and a timestamp/version field. Filter the stream for the latest ID/version field.
@@ -197,4 +224,4 @@ groups:
       args:
         - debug
         - warn
-        - quite
+        - quite

src/cli.yaml

@@ -2,7 +2,6 @@ use std::fs::File;
 use std::io::Write;
 
 use comrak::{markdown_to_html, ComrakOptions};
-use mime::{Mime, APPLICATION_JSON, APPLICATION_OCTET_STREAM, TEXT_JAVASCRIPT};
 use serde_json;
 
 use super::model::*;
@@ -128,25 +127,16 @@ impl ToMarkdown for HarFile {
                 entry.response.content.mime_type.clone()
             ));
             if let Some(text) = entry.response.content.text {
-                let body = text.replace("\\n", "\n");
-                let mime = entry
-                    .response
-                    .content
-                    .mime_type
-                    .parse::<Mime>()
-                    .unwrap_or(APPLICATION_OCTET_STREAM);
-
                 lines.push("**Body:**".to_string());
-                if mime == APPLICATION_JSON || mime == TEXT_JAVASCRIPT {
-                    match serde_json::from_str::<serde_json::Value>(&body) {
-                        Ok(json) => {
-                            let body = serde_json::to_string_pretty(&json).unwrap();
-                            lines.push(format!("```\n{}\n```\n", body))
-                        }
-                        Err(_) => lines.push(format!("```\n{}\n```\n", body)),
+                match serde_json::from_str::<serde_json::Value>(&text) {
+                    Ok(json) => {
+                        let body = serde_json::to_string_pretty(&json).unwrap();
+                        lines.push(format!("```\n{}\n```\n", body))
+                    }
+                    Err(_) => {
+                        let body = text.replace("\\n", "\n");
+                        lines.push(format!("```\n{}\n```\n", body));
                     }
-                } else {
-                    lines.push(format!("```\n{}\n```\n", body))
                 }
             }
         }

src/commands/har/output.rs

@@ -0,0 +1,140 @@
+use std::collections::BTreeMap;
+use std::fs::File;
+use std::io::prelude::*;
+use std::io::{BufRead, BufReader, LineWriter};
+use std::path::PathBuf;
+
+use clap::ArgMatches;
+use json::JsonValue;
+
+use super::{find_field, parse_path};
+use crate::commands::progress::*;
+use crate::commands::CliError;
+
+#[derive(Debug)]
+struct Record {
+    id: String,
+    version: i32,
+    data: String,
+}
+
+#[test]
+fn leading_dot_will_be_ignored() {
+    let split: Vec<String> = parse_path(".abc.123");
+
+    assert_eq!(2, split.len());
+}
+
+fn build_key(keys: &Vec<Vec<String>>, json_input: &JsonValue) -> Option<String> {
+    let mut key_list: Vec<String> = Vec::new();
+
+    for part in keys {
+        if let Some(key) = find_field(part, json_input) {
+            key_list.push(s!(key.as_str().unwrap_or("null")));
+        }
+    }
+
+    if key_list.is_empty() {
+        return None;
+    }
+
+    return Some(key_list.join(":"));
+}
+
+pub fn do_json_latest_command(args: &ArgMatches) -> Result<(), CliError> {
+    let output_path = args.value_of("OUTPUT").unwrap();
+    let output_path = PathBuf::from(output_path);
+
+    let id_fields: Vec<Vec<String>> = args
+        .values_of("id")
+        .unwrap()
+        .into_iter()
+        .map(|x| parse_path(x))
+        .collect();
+
+    let version_path: Vec<String> = parse_path(args.value_of("seq").unwrap());
+
+    let file = File::create(output_path).unwrap();
+    let mut file = LineWriter::new(file);
+
+    let pb = ProgressBarHelper::new(ProgressBarType::UnsizedProgressBar(
+        "{prefix:.bold.dim} {spinner:.green} {wide_msg}",
+    ));
+
+    let mut records: BTreeMap<String, Record> = BTreeMap::new();
+
+    let input_paths: Vec<String> = args.values_of("INPUT").unwrap().map(|x| s!(x)).collect();
+
+    let mut counter = 0;
+    for input_path in input_paths.into_iter() {
+        let reader = match crate::commands::file::open_file(&input_path) {
+            Ok(reader) => BufReader::new(reader),
+            Err(e) => {
+                error!("Unable to open {} because {}", input_path, e.to_string());
+                continue;
+            }
+        };
+
+        for line in reader.lines() {
+            match line {
+                Ok(line) => {
+                    counter += 1;
+                    if counter % 10 == 0 {
+                        pb.set_message(&format!(
+                            "Reading line {}\t Used: {}",
+                            counter,
+                            records.len()
+                        ));
+                    }
+
+                    let json_line = json::parse(&line).unwrap();
+
+                    let id = match build_key(&id_fields, &json_line) {
+                        Some(value) => value,
+                        None => {
+                            warn!("Skipping `{}` because all id was missing.", &line);
+                            continue;
+                        }
+                    };
+
+                    let version = match find_field(&version_path, &json_line) {
+                        Some(value) => value.as_i32().unwrap(),
+                        None => {
+                            warn!("Skipping `{}` because version was missing.", &line);
+                            continue;
+                        }
+                    };
+
+                    let record = Record {
+                        id: id.to_string(),
+                        version: version,
+                        data: line,
+                    };
+
+                    match records.get(&id) {
+                        Some(row) if row.version > version => continue,
+                        _ => records.insert(id.to_string(), record),
+                    };
+                }
+                Err(err) => error!("IO error: Line {}: {}", counter, err),
+            }
+        }
+    }
+
+    let mut write_counter = 0;
+
+    for record in records.values() {
+        write_counter += 1;
+        pb.set_message(&format!(
+            "Writing line {} of {}",
+            write_counter,
+            records.len()
+        ));
+        let line = format!("{}\n", record.data);
+        if file.write_all(line.as_bytes()).is_err() {
+            error!("Trouble writing line {} to disk", write_counter);
+        }
+    }
+
+    return Ok(());
+}

src/commands/json/latest.rs

@@ -1,27 +1,24 @@
-use std::collections::BTreeMap;
-use std::fs::File;
-use std::io::prelude::*;
-use std::io::{BufRead, BufReader, LineWriter};
-use std::path::PathBuf;
+mod latest;
+mod sql;
+
+pub use self::latest::do_json_latest_command;
+pub use self::sql::do_json_sql_command;
 
-use clap::ArgMatches;
-use indicatif::{ProgressBar, ProgressStyle};
 use json::JsonValue;
 
-use crate::commands::CliError;
+// https://github.com/ms705/nom-sql
 
-#[derive(Debug)]
-struct Record {
-    id: String,
-    version: i32,
-    data: String,
-}
+fn find_field<'a>(field: &Vec<String>, json_input: &'a JsonValue) -> Option<&'a JsonValue> {
+    let mut value = json_input;
 
-#[test]
-fn leading_dot_will_be_ignored() {
-    let split: Vec<String> = parse_path(".abc.123");
+    for part in field {
+        value = &value[part];
+        if value.is_null() {
+            return None;
+        }
+    }
 
-    assert_eq!(2, split.len());
+    return Some(value);
 }
 
 fn parse_path(path: &str) -> Vec<String> {
@@ -35,132 +32,3 @@ fn parse_path(path: &str) -> Vec<String> {
 
     return split;
 }
-
-fn build_key(keys: &Vec<Vec<String>>, json_input: &JsonValue) -> Option<String> {
-    let mut key_list: Vec<String> = Vec::new();
-
-    for part in keys {
-        if let Some(key) = find_field(part, json_input) {
-            key_list.push(s!(key.as_str().unwrap_or("null")));
-        }
-    }
-
-    if key_list.is_empty() {
-        return None;
-    }
-
-    return Some(key_list.join(":"));
-}
-
-pub fn do_json_latest_command(args: &ArgMatches) -> Result<(), CliError> {
-    let output_path = args.value_of("OUTPUT").unwrap();
-    let output_path = PathBuf::from(output_path);
-
-    let id_fields: Vec<Vec<String>> = args
-        .values_of("id")
-        .unwrap()
-        .into_iter()
-        .map(|x| parse_path(x))
-        .collect();
-
-    let version_path: Vec<String> = parse_path(args.value_of("seq").unwrap());
-
-    let file = File::create(output_path).unwrap();
-    let mut file = LineWriter::new(file);
-
-    let spinner_style =
-        ProgressStyle::default_spinner().template("{prefix:.bold.dim} {spinner:.green} {wide_msg}");
-    let progress_bar = ProgressBar::new_spinner();
-    progress_bar.set_style(spinner_style.clone());
-    progress_bar.enable_steady_tick(100);
-
-    let mut records: BTreeMap<String, Record> = BTreeMap::new();
-
-    let input_paths: Vec<String> = args.values_of("INPUT").unwrap().map(|x| s!(x)).collect();
-
-    let mut counter = 0;
-    for input_path in input_paths.into_iter() {
-        let reader = match crate::commands::file::open_file(&input_path) {
-            Ok(reader) => BufReader::new(reader),
-            Err(e) => {
-                error!("Unable to open {} because {}", input_path, e.to_string());
-                continue;
-            }
-        };
-
-        for line in reader.lines() {
-            match line {
-                Ok(line) => {
-                    counter += 1;
-                    if counter % 10 == 0 {
-                        progress_bar.set_message(&format!(
-                            "Reading line {}\t Used: {}",
-                            counter,
-                            records.len()
-                        ));
-                    }
-
-                    let json_line = json::parse(&line).unwrap();
-
-                    let id = match build_key(&id_fields, &json_line) {
-                        Some(value) => value,
-                        None => {
-                            warn!("Skipping `{}` because all id was missing.", &line);
-                            continue;
-                        }
-                    };
-
-                    let version = match find_field(&version_path, &json_line) {
-                        Some(value) => value.as_i32().unwrap(),
-                        None => {
-                            warn!("Skipping `{}` because version was missing.", &line);
-                            continue;
-                        }
-                    };
-
-                    let record = Record {
-                        id: id.to_string(),
-                        version: version,
-                        data: line,
-                    };
-
-                    match records.get(&id) {
-                        Some(row) if row.version > version => continue,
-                        _ => records.insert(id.to_string(), record),
-                    };
-                }
-                Err(err) => error!("IO error: Line {}: {}", counter, err),
-            }
-        }
-    }
-
-    let mut write_counter = 0;
-
-    for record in records.values() {
-        write_counter += 1;
-        progress_bar.set_message(&format!(
-            "Writing line {} of {}",
-            write_counter,
-            records.len()
-        ));
-        let line = format!("{}\n", record.data);
-        if file.write_all(line.as_bytes()).is_err() {
-            error!("Trouble writing line {} to disk", write_counter);
-        }
-    }
-
-    return Ok(());
-}
-
-fn find_field<'a>(field: &Vec<String>, json_input: &'a JsonValue) -> Option<&'a JsonValue> {
-    let mut value = json_input;
-
-    for part in field {
-        value = &value[part];
-        if value.is_null() {
-            return None;
-        }
-    }
-
-    return Some(value);
-}

src/commands/json/mod.rs

@@ -0,0 +1,103 @@
+mod parser;
+
+use std::io::{BufRead, BufReader, Read};
+
+use clap::ArgMatches;
+use json::JsonValue;
+use parser::{Expression, SqlSource};
+
+use super::{find_field, parse_path};
+
+use crate::commands::CliError;
+
+pub fn do_json_sql_command(args: &ArgMatches) -> Result<(), CliError> {
+    let exp = Expression::from(args.value_of("EXP").unwrap())?;
+
+    let from_path = match exp.source {
+        SqlSource::None => parse_path("."),
+        SqlSource::SqlFrom { path } => parse_path(&path),
+    };
+
+    let input_paths: Vec<String> = args.values_of("json").unwrap().map(|x| s!(x)).collect();
+
+    let line_processor = LineProcessor { path: from_path };
+
+    for input_path in input_paths.into_iter() {
+        let reader = match crate::commands::file::open_file(&input_path) {
+            Ok(reader) => BufReader::new(reader),
+            Err(e) => {
+                error!("Unable to open {} because {}", input_path, e.to_string());
+                continue;
+            }
+        };
+
+        line_processor.process_stream(reader);
+    }
+
+    Ok(())
+}
+
+struct LineProcessor {
+    path: Vec<String>,
+}
+
+impl LineProcessor {
+    fn process_stream<R: Read>(&self, buf_read: BufReader<R>) {
+        let mut line_counter = 0;
+        for line in buf_read.lines() {
+            line_counter += 1;
+            match line {
+                Ok(line) => {
+                    self.parse_json_line(line, line_counter);
+                }
+                Err(err) => error!("IO error: Line {}: {}", line_counter, err),
+            }
+        }
+    }
+
+    fn parse_json_line(&self, line: String, line_number: u32) {
+        match json::parse(&line) {
+            Ok(parsed) => {
+                if let Some(sub_json) = find_field(&self.path, &parsed) {
+                    trace!("sub json: {:?}", sub_json);
+                    self.process_json_line(sub_json)
+                }
+            }
+            Err(_) => error!("Line was not JSON: {}", line_number),
+        }
+    }
+
+    fn process_json_line(&self, parent: &JsonValue) {
+        match parent {
+            JsonValue::Array(values) => {
+                for value in values {
+                    if self.filter_obj(&value) {
+                        self.print_fields(&value);
+                    }
+                }
+            }
+            JsonValue::Object(_) => {
+                if self.filter_obj(parent) {
+                    self.print_fields(&parent);
+                }
+            }
+            _ => error!(
+                "Tried to parse an object that wasn't an array or object, got {:?}",
+                parent
+            ),
+        }
+    }
+
+    fn filter_obj(&self, parent: &JsonValue) -> bool {
+        if let JsonValue::Object(_obj) = parent {
+            true
+        } else {
+            error!("{:?} is not a json object", parent);
+            false
+        }
+    }
+
+    fn print_fields(&self, parent: &JsonValue) {
+        println!("{}", parent.dump());
+    }
+}

src/commands/json/sql/mod.rs

@@ -0,0 +1,268 @@
+use pest::iterators::{Pair, Pairs};
+use pest::Parser;
+
+use crate::commands::CliError;
+#[cfg(test)]
+use pest::{consumes_to, parses_to};
+
+#[derive(Parser)]
+#[grammar = "commands/json/sql/sql.pest"] // relative to src
+struct SqlParser;
+
+#[derive(PartialEq, Debug)]
+pub struct Expression {
+    pub operation: SqlOperation,
+    pub source: SqlSource,
+    pub filter: Vec<SqlFilter>,
+    pub limit: Option<u32>,
+    pub offset: Option<u32>,
+}
+
+impl Expression {
+    fn new() -> Self {
+        Expression {
+            operation: SqlOperation::None,
+            source: SqlSource::None,
+            filter: Vec::new(),
+            limit: None,
+            offset: None,
+        }
+    }
+
+    pub fn from(sql: &str) -> Result<Self, CliError> {
+        let mut expression = Expression::new();
+        let mut pairs = match SqlParser::parse(Rule::bool_expr, sql) {
+            Ok(pairs) => pairs,
+            Err(e) => {
+                return Err(CliError::new(format!("{}", e), 5));
+            }
+        };
+
+        extract(&mut expression, &mut pairs);
+
+        Ok(expression)
+    }
+}
+
+#[derive(PartialEq, Debug)]
+pub enum SqlOperation {
+    None,
+    Select { columns: Vec<String> },
+}
+
+#[derive(PartialEq, Debug)]
+pub enum SqlSource {
+    None,
+    SqlFrom { path: String },
+}
+
+#[derive(PartialEq, Debug)]
+pub enum SqlFilter {
+    Condition(SqlComparison),
+    Or,
+    And,
+}
+
+#[derive(PartialEq, Debug)]
+pub enum SqlComparison {
+    Equal { path: String, value: String },
+    NotEqual { path: String, value: String },
+}
+
+fn extract(expression: &mut Expression, bool_expr: &mut Pairs<Rule>) {
+    let mut exp = bool_expr.next().unwrap().into_inner();
+    let pairs = exp.next().unwrap().into_inner();
+    for pair in pairs {
+        trace!("Element: {:?}", pair);
+        match pair.as_rule() {
+            Rule::select_operation => {
+                handle_select(expression, pair);
+            }
+            Rule::source => {
+                handle_from(expression, pair);
+            }
+            Rule::filter => {
+                handle_where(expression, pair);
+            }
+            Rule::limit => {}
+            Rule::offset => {}
+            _ => panic!("SQL Parse error! {:?}", pair),
+        }
+    }
+}
+
+fn handle_select(expression: &mut Expression, parent: Pair<Rule>) {
+    let inner = parent.into_inner();
+
+    let columns: Vec<String> = inner.map(|x| s!(x.as_span().as_str())).collect();
+    expression.operation = SqlOperation::Select { columns: columns };
+}
+
+fn handle_from(expression: &mut Expression, parent: Pair<Rule>) {
+    expression.source = SqlSource::SqlFrom {
+        path: s!(parent.into_inner().as_str()),
+    };
+}
+
+fn handle_where(expression: &mut Expression, parent: Pair<Rule>) {
+    let mut filters: Vec<SqlFilter> = Vec::new();
+
+    for pair in parent.into_inner() {
+        match pair.as_rule() {
+            Rule::condition => {
+                filters.push(handle_condition(&mut pair.into_inner()));
+            }
+            Rule::logic => match pair.into_inner().next().unwrap().as_rule() {
+                Rule::and => {
+                    filters.push(SqlFilter::And);
+                }
+                Rule::or => {
+                    filters.push(SqlFilter::Or);
+                }
+                _ => {
+                    panic!("Unknown operator");
+                }
+            },
+            _ => panic!("Unable to parse where"),
+        }
+    }
+
+    expression.filter = filters;
+}
+
+fn handle_condition(parent: &mut Pairs<Rule>) -> SqlFilter {
+    let path = s!(parent.next().unwrap().as_str());
+    let operator = parent.next().unwrap();
+    let value = s!(parent.next().unwrap().as_str());
+
+    let comparison = match operator.as_rule() {
+        Rule::eq => SqlComparison::Equal {
+            path: path,
+            value: value,
+        },
+        Rule::neq => SqlComparison::NotEqual {
+            path: path,
+            value: value,
+        },
+        _ => panic!(
+            "Value must be `=` or `!=`, but was {:?} ({:?})",
+            operator.as_str(),
+            operator.as_rule()
+        ),
+    };
+
+    SqlFilter::Condition(comparison)
+}
+
+#[test]
+fn validate_ast_builder() {
+    let mut pairs =
+        SqlParser::parse(Rule::bool_expr, "select * from . where .a.b.c = 123").unwrap();
+    let mut expression = Expression::new();
+
+    extract(&mut expression, &mut pairs);
+
+    assert_eq!(
+        SqlOperation::Select {
+            columns: vec![s!("*")]
+        },
+        expression.operation
+    );
+    assert_eq!(SqlSource::SqlFrom { path: s!(".") }, expression.source);
+    assert_eq!(
+        vec![SqlFilter::Condition(SqlComparison::Equal {
+            path: s!(".a.b.c"),
+            value: s!("123")
+        })],
+        expression.filter
+    );
+}
+
+#[test]
+fn validate_parsing() {
+    parses_to! {
+        parser: SqlParser,
+        input: "select * from . where .a.b.c = 123",
+        rule: Rule::expr,
+        tokens: [
+            expr(0, 34, [
+                select_operation(0,9,[
+                    result_column(7,8, [])
+                ]),
+                source(9, 15, [
+                    json_path(14, 15, []),
+                ]),
+                filter(16, 34, [
+                    condition(22, 34, [
+                        json_path(22, 28, []),
+                        eq(29, 30, []),
+                        value(31, 34, [num_literal(31, 34, [])])
+                    ])
+                ])
+            ])
+        ]
+    }
+
+    parses_to! {
+        parser: SqlParser,
+        input: "select .abc from .a.b.c",
+        rule: Rule::expr,
+        tokens: [
+            expr(0, 23, [
+                select_operation(0, 12,[
+                    result_column(7, 11, [
+                        json_path(7, 11, [])
+                    ])
+                ]),
+                source(12, 23, [
+                    json_path(17, 23, []),
+                ])
+            ])
+        ]
+    }
+
+    parses_to! {
+        parser: SqlParser,
+        input: "select .abc from .a.b.c limit 10 offset 10",
+        rule: Rule::expr,
+        tokens: [
+            expr(0, 42, [
+                select_operation(0, 12,[
+                    result_column(7, 11, [
+                        json_path(7, 11, [])
+                    ])
+                ]),
+                source(12, 23, [
+                    json_path(17, 23, []),
+                ]),
+                limit(24, 32, [
+                    pos_num_literal(30, 32, []),
+                ]),
+                offset(33, 42, [
+                    pos_num_literal(40, 42, []),
+                ])
+            ])
+        ]
+    }
+
+    parses_to! {
+        parser: SqlParser,
+        input: "select .abc from .a.b.c limit 10",
+        rule: Rule::expr,
+        tokens: [
+            expr(0, 32, [
+                select_operation(0, 12,[
+                    result_column(7, 11, [
+                        json_path(7, 11, [])
+                    ])
+                ]),
+                source(12, 23, [
+                    json_path(17, 23, []),
+                ]),
+                limit(24, 32, [
+                    pos_num_literal(30, 32, []),
+                ])
+            ])
+        ]
+    }
+}

src/commands/json/sql/parser.rs

@@ -0,0 +1,47 @@
+json_path = ${ ("." ~ (ASCII_ALPHANUMERIC){1,}){1,} | "." } 
+
+bool_expr = { SOI ~ expr ~ EOI }
+expr = { operation ~ source ~ filter? ~ (limit ~ offset?)? }
+
+operation = _{ select_operation }
+source = { ^"FROM" ~ json_path }
+filter = { ^"WHERE" ~ condition ~ ( logic ~ condition )* }
+
+logic = { and | or }
+and = { ^"and" }
+or = { ^"or" }
+
+condition = { json_path ~ comparison ~ value }
+
+limit = { ^"LIMIT" ~ pos_num_literal }
+offset = { ^"OFFSET" ~ pos_num_literal }
+
+result_column = { "*" | json_path }
+select_operation = { ^"SELECT" ~ result_column ~ ("," ~ result_column )* }
+
+comparison = _{ eq | neq }
+eq = { "=" }
+neq = { "!=" | "<>"}
+
+value = {
+    string_literal
+    | num_literal
+    | "(" ~ string_literal ~("," ~ string_literal)* ~ ")"
+    | "(" ~ num_literal ~("," ~ num_literal)* ~ ")"
+}
+
+pos_num_literal = @{ ("0" | ASCII_NONZERO_DIGIT ~ ASCII_DIGIT*) }
+
+num_literal = @{
+    "-"?
+    ~ ("0" | ASCII_NONZERO_DIGIT ~ ASCII_DIGIT*)
+    ~ ("." ~ ASCII_DIGIT*)?
+}
+
+string_literal = ${ "\"" ~ string ~ "\"" }
+string = @{ char* }
+char = {
+    !("\"" | "\\") ~ ANY
+}
+
+WHITESPACE = _{ " " }

src/commands/json/sql/sql.pest

@@ -31,4 +31,5 @@ mod file;
 pub mod har;
 pub mod json;
 pub mod nsq;
+pub mod progress;
 pub mod time;

src/commands/mod.rs

@@ -1,3 +1,4 @@
 pub mod api;
+pub mod model;
 pub mod post;
 pub mod stats;

src/commands/nsq/api.rs

@@ -0,0 +1,208 @@
+#[derive(Serialize, Deserialize)]
+pub struct StatusTopicsResponse {
+    pub data: StatusTopicsDetails,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct StatusTopicsDetails {
+    pub topics: Vec<TopicDetails>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct TopicDetails {
+    pub topic_name: String,
+    pub depth: u64,
+    pub message_count: u64,
+    pub channels: Vec<TopicChannel>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct TopicChannel {
+    pub depth: u64,
+    pub in_flight_count: u64,
+    pub channel_name: String,
+    pub message_count: u64,
+    pub clients: Vec<ClientDetails>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct ClientDetails {
+    pub hostname: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct LookupProducer {
+    pub remote_address: String,
+    pub hostname: String,
+    pub broadcast_address: String,
+    pub tcp_port: i32,
+    pub http_port: i32,
+    pub version: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct LookupResponse {
+    pub status_code: i32,
+    pub data: LookupData,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct LookupData {
+    pub producers: Vec<LookupProducer>,
+}
+
+#[test]
+fn test_extract_size() {
+    let body = "{
+    \"version\": \"1.1.0\",
+    \"health\": \"OK\",
+    \"start_time\": 1548185315,
+    \"topics\": [
+        {
+            \"topic_name\": \"foo\",
+            \"channels\": [
+                {
+                    \"channel_name\": \"tail180292#ephemeral\",
+                    \"depth\": 3,
+                    \"backend_depth\": 0,
+                    \"in_flight_count\": 1,
+                    \"deferred_count\": 0,
+                    \"message_count\": 1399,
+                    \"requeue_count\": 0,
+                    \"timeout_count\": 0,
+                    \"clients\": [
+                        {
+                            \"client_id\": \"ethan\",
+                            \"hostname\": \"ethan.local\",
+                            \"version\": \"V2\",
+                            \"remote_address\": \"1.2.3.4:33576\",
+                            \"state\": 3,
+                            \"ready_count\": 1,
+                            \"in_flight_count\": 1,
+                            \"message_count\": 1396,
+                            \"finish_count\": 1395,
+                            \"requeue_count\": 0,
+                            \"connect_ts\": 1549065745,
+                            \"sample_rate\": 0,
+                            \"deflate\": false,
+                            \"snappy\": false,
+                            \"user_agent\": \"nsq_tail/1.1.0 go-nsq/1.0.6\",
+                            \"tls\": false,
+                            \"tls_cipher_suite\": \"\",
+                            \"tls_version\": \"\",
+                            \"tls_negotiated_protocol\": \"\",
+                            \"tls_negotiated_protocol_is_mutual\": false
+                        }
+                    ],
+                    \"paused\": false,
+                    \"e2e_processing_latency\": {
+                        \"count\": 0,
+                        \"percentiles\": null
+                    }
+                }
+            ],
+            \"depth\": 0,
+            \"backend_depth\": 0,
+            \"message_count\": 29259,
+            \"paused\": false,
+            \"e2e_processing_latency\": {
+                \"count\": 0,
+                \"percentiles\": null
+            }
+        }
+    ],
+    \"memory\": {
+        \"heap_objects\": 21625,
+        \"heap_idle_bytes\": 11886592,
+        \"heap_in_use_bytes\": 3743744,
+        \"heap_released_bytes\": 10280960,
+        \"gc_pause_usec_100\": 5612,
+        \"gc_pause_usec_99\": 3742,
+        \"gc_pause_usec_95\": 878,
+        \"next_gc_bytes\": 4194304,
+        \"gc_total_runs\": 219
+    }
+}";
+
+    serde_json::from_str::<StatusTopicsDetails>(body).unwrap();
+}
+
+#[test]
+fn older_api_test() {
+    let body = "{
+  \"status_code\": 200,
+  \"status_txt\": \"OK\",
+  \"data\": {
+    \"version\": \"0.3.8\",
+    \"health\": \"OK\",
+    \"start_time\": 1543350728,
+    \"topics\": [
+        {
+            \"topic_name\": \"foo\",
+            \"channels\": [
+                {
+                    \"channel_name\": \"tail180292#ephemeral\",
+                    \"depth\": 3,
+                    \"backend_depth\": 0,
+                    \"in_flight_count\": 1,
+                    \"deferred_count\": 0,
+                    \"message_count\": 1399,
+                    \"requeue_count\": 0,
+                    \"timeout_count\": 0,
+                    \"clients\": [
+                        {
+                            \"client_id\": \"ethan\",
+                            \"hostname\": \"ethan.local\",
+                            \"version\": \"V2\",
+                            \"remote_address\": \"1.2.3.4:33576\",
+                            \"state\": 3,
+                            \"ready_count\": 1,
+                            \"in_flight_count\": 1,
+                            \"message_count\": 1396,
+                            \"finish_count\": 1395,
+                            \"requeue_count\": 0,
+                            \"connect_ts\": 1549065745,
+                            \"sample_rate\": 0,
+                            \"deflate\": false,
+                            \"snappy\": false,
+                            \"user_agent\": \"nsq_tail/1.1.0 go-nsq/1.0.6\",
+                            \"tls\": false,
+                            \"tls_cipher_suite\": \"\",
+                            \"tls_version\": \"\",
+                            \"tls_negotiated_protocol\": \"\",
+                            \"tls_negotiated_protocol_is_mutual\": false
+                        }
+                    ],
+                    \"paused\": false,
+                    \"e2e_processing_latency\": {
+                        \"count\": 0,
+                        \"percentiles\": null
+                    }
+                }
+            ],
+            \"depth\": 0,
+            \"backend_depth\": 0,
+            \"message_count\": 29259,
+            \"paused\": false,
+            \"e2e_processing_latency\": {
+                \"count\": 0,
+                \"percentiles\": null
+            }
+        }
+    ],
+    \"memory\": {
+        \"heap_objects\": 21625,
+        \"heap_idle_bytes\": 11886592,
+        \"heap_in_use_bytes\": 3743744,
+        \"heap_released_bytes\": 10280960,
+        \"gc_pause_usec_100\": 5612,
+        \"gc_pause_usec_99\": 3742,
+        \"gc_pause_usec_95\": 878,
+        \"next_gc_bytes\": 4194304,
+        \"gc_total_runs\": 219
+        }
+    }
+}";
+
+    serde_json::from_str::<StatusTopicsResponse>(body).unwrap();
+}

src/commands/nsq/mod.rs

@@ -8,8 +8,8 @@ use clap::ArgMatches;
 use crossbeam_channel::bounded;
 use crossbeam_channel::Receiver;
 
-use indicatif::{ProgressBar, ProgressStyle};
-
+use crate::commands::nsq::api::*;
+use crate::commands::progress::*;
 use crate::commands::CliError;
 
 const RATE_LIMIT: &str = "200";
@@ -107,6 +107,13 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
         )
     };
 
+    let status = NsqState::new(
+        &options.nsq_lookup,
+        NsqFilter::Topic {
+            topics: vec![options.topic.clone()].into_iter().collect(),
+        },
+    );
+
     debug!("Capacity of in messages: {}", capacity);
     debug!("Interval of new tokens: {:?}", interval);
 
@@ -117,21 +124,20 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
 
     THREADS_RUNNING.store(true, Ordering::SeqCst);
 
-    let urls = super::api::get_base_url_for_topic(&options.nsq_lookup, &options.topic);
-    let base_url = if urls.is_empty() {
-        return Err(CliError::new("Unable to get NSQ Host", 2));
-    } else {
-        format!("{}", urls.first().unwrap())
+    let base_addresss = match status.get_topic_url(&options.topic.clone()) {
+        Some(address) => address,
+        None => {
+            error!("NSQ does now know about topic {}", options.topic);
+            return Err(CliError::new("Unable to get NSQ Host", 2));
+        }
     };
 
-    let submit_url = format!("{}/pub?topic={}", base_url, &options.topic);
+    let submit_url = format!("{}/pub?topic={}", base_addresss, &options.topic);
 
-    let style = ProgressStyle::default_bar()
-        .template("[{elapsed_precise}] {bar:80.cyan/blue} {pos:>7}/{len:7} {msg}")
-        .progress_chars("##-");
-    let progress_bar = ProgressBar::new(options.limit as u64);
-    progress_bar.set_style(style.clone());
-    progress_bar.enable_steady_tick(1000);
+    let pb = ProgressBarHelper::new(ProgressBarType::SizedProgressBar(
+        options.limit,
+        "[{elapsed_precise}] {bar:80.cyan/blue} {pos:>7}/{len:7} {msg}",
+    ));
 
     let (s1, r1) = bounded(20);
 
@@ -146,8 +152,8 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
     }
 
     let topic = format!("{}", options.topic);
-    do_api_check(&base_url, &topic);
-    threads.push(thread::spawn(move || check_api_status(&base_url, &topic)));
+    do_api_check(&topic, &status);
+    threads.push(thread::spawn(move || check_api_status(&topic, &status)));
 
     let reader = crate::commands::file::open_file(options.file.to_str().unwrap())?;
     let reader = BufReader::new(reader);
@@ -157,7 +163,7 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
         loop {
             let max_depth = API_DEPTH.load(Ordering::SeqCst);
             let in_flight = API_IN_FLIGHT.load(Ordering::SeqCst);
-            progress_bar.set_message(&format!(
+            pb.set_message(&format!(
                 "In Progress: {:4}\tBacklog Size: {:4}\tOffset: {}",
                 in_flight,
                 max_depth,
@@ -178,7 +184,7 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
         }
 
         ratelimit.wait();
-        progress_bar.inc(1);
+        pb.inc();
 
         if options.offset > counter {
             OFFSET.fetch_add(1, Ordering::SeqCst);
@@ -189,7 +195,7 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
             ERRORS.fetch_add(1, Ordering::SeqCst);
         }
     }
-    progress_bar.finish();
+    pb.done();
 
     THREADS_RUNNING.store(false, Ordering::SeqCst);
 
@@ -211,24 +217,22 @@ pub fn do_send_command(args: &ArgMatches) -> Result<(), CliError> {
     return Ok(());
 }
 
-fn check_api_status(base_url: &str, topic: &str) {
+fn check_api_status(topic: &str, state: &NsqState) {
     loop {
         if !THREADS_RUNNING.load(Ordering::SeqCst) {
             return;
         }
 
-        do_api_check(base_url, topic);
+        do_api_check(topic, state);
+        std::thread::sleep(Duration::from_millis(200));
     }
 }
 
-fn do_api_check(base_url: &str, topic: &str) {
-    if let Some((max_depth, in_flight)) = super::api::get_queue_size(base_url, topic) {
-        let max_depth = std::cmp::max(0, max_depth) as usize;
-        let in_flight = std::cmp::max(0, in_flight) as usize;
-        API_IN_FLIGHT.store(in_flight, Ordering::SeqCst);
-        API_DEPTH.store(max_depth, Ordering::SeqCst);
-        std::thread::sleep(Duration::from_millis(200));
-    }
+fn do_api_check(topic: &str, state: &NsqState) {
+    let snapshot = state.update_status();
+    let agg = snapshot.topics.get(topic).unwrap().producer_aggregate();
+    let max_depth = std::cmp::max(0, agg.depth) as usize;
+    API_DEPTH.store(max_depth, Ordering::SeqCst);
 }
 
 fn process_messages(reciever: Receiver<String>, path: String) {

src/commands/nsq/model.rs

@@ -0,0 +1,56 @@
+use indicatif::{ProgressBar, ProgressStyle};
+
+pub struct ProgressBarHelper {
+    pb: ProgressBar,
+}
+
+pub enum ProgressBarType<'a> {
+    SizedProgressBar(usize, &'a str),
+    UnsizedProgressBar(&'a str),
+}
+
+impl ProgressBarHelper {
+    pub fn new(p_type: ProgressBarType) -> Self {
+        if atty::isnt(atty::Stream::Stdout) {
+            ProgressBarHelper {
+                pb: ProgressBar::hidden(),
+            }
+        } else {
+            let template = match &p_type {
+                ProgressBarType::SizedProgressBar(_, template) => template,
+                ProgressBarType::UnsizedProgressBar(template) => template,
+            };
+
+            let pb = match p_type {
+                ProgressBarType::SizedProgressBar(size, _) => ProgressBar::new(size as u64),
+                ProgressBarType::UnsizedProgressBar(_) => ProgressBar::new_spinner(),
+            };
+
+            let spinner_style = ProgressStyle::default_spinner()
+                .tick_chars("⠁⠂⠄⡀⢀⠠⠐⠈ ")
+                .progress_chars("#>-")
+                .template(&template);
+            pb.set_style(spinner_style.clone());
+            pb.enable_steady_tick(100);
+            ProgressBarHelper { pb }
+        }
+    }
+
+    #[allow(dead_code)]
+    pub fn inc_with_message(&self, message: &str) {
+        self.pb.inc(1);
+        self.pb.set_message(message);
+    }
+
+    pub fn inc(&self) {
+        self.pb.inc(1);
+    }
+
+    pub fn set_message(&self, message: &str) {
+        self.pb.set_message(message);
+    }
+
+    pub fn done(&self) {
+        self.pb.finish_and_clear();
+    }
+}

src/commands/nsq/post.rs

@@ -4,40 +4,29 @@ mod parse;
 
 use clap::ArgMatches;
 
-use colored::*;
 use chrono::prelude::*;
+use colored::*;
 
-use parse::TimeResult;
 use crate::commands::CliError;
+use parse::TimeResult;
 
 pub fn do_time_command(args: &ArgMatches) -> Result<(), CliError> {
     let input_array: Vec<&str> = args.values_of("INPUT").unwrap().collect();
     let input_string = input_array.join(" ");
-    return match (
-        parse::parse(&input_string),
-        args.is_present("utc_only"),
-    ) {
+    return match (parse::parse(&input_string), args.is_present("utc_only")) {
         (Ok(date), true) => render_utc(date),
         (Ok(date), false) => render_full_output(date),
         (Err(_), _) => {
-                error!(
-                    "Unable to understand `{}`",
-                    input_string
-                );
-                return Err(CliError::new("Unknown format", 1));
-            }
+            error!("Unable to understand `{}`", input_string);
+            return Err(CliError::new("Unknown format", 1));
+        }
     };
 }
 
-
 fn render_full_output(input: TimeResult) -> Result<(), CliError> {
     let datetimes = match input {
-        TimeResult::Epoch(epoch) => {
-            epoch.make_permutations()
-        },
-        TimeResult::String(string_format) => {
-            string_format.make_permutations()
-        }
+        TimeResult::Epoch(epoch) => epoch.make_permutations(),
+        TimeResult::String(string_format) => string_format.make_permutations(),
     };
 
     let mut first = true;
@@ -48,35 +37,96 @@ fn render_full_output(input: TimeResult) -> Result<(), CliError> {
         first = false;
 
         println!("Understood the date was {}", format!("{}", datetime).bold());
-        println!("  ├── Date 'human': {}", format!("{}", datetime.format("%b %e %T %Y")).bold());
-        println!("  ├── Date in M/D/Y: {}", format!("{}/{}/{}", datetime.month(), datetime.day(), datetime.year()).bold());
-        println!("  ├── Date in YMD: {}", format!("{}{:02}{:02}", datetime.year(), datetime.month(), datetime.day()).bold());
-        println!("  ├── Day in year: {}", format!("{}", datetime.ordinal()).bold());
-        println!("  ├── ISO week {}", format!("{}-{}", datetime.iso_week().year(), datetime.iso_week().week()).bold());
-        println!("  ├── Day of week: {}", format!("{:?}", datetime.weekday()).bold());
-        println!("  └── Time: {}", format!("{} {}", datetime.time(), datetime.timezone()).bold());
-        println!("    ├── Unix epoch(s):  {}", format!("{}", datetime.timestamp()).bold());
-        println!("    ├── Unix epoch(ms): {}", format!("{}", datetime.timestamp_millis()).bold());
-        println!("    ├── Unix epoch(ns): {}", format!("{}", datetime.timestamp_nanos()).bold());
-        println!("    ├── In UTC: {}", format!("{}", datetime.with_timezone(&chrono::Utc)).bold());
-        println!("    ├── In Eastern: {}", format!("{}", datetime.with_timezone(&chrono_tz::US::Eastern)).bold());
-        println!("    ├── In Central: {}", format!("{}", datetime.with_timezone(&chrono_tz::US::Central)).bold());
-        println!("    ├── In Mountain: {}", format!("{}", datetime.with_timezone(&chrono_tz::US::Mountain)).bold());
-        println!("    ├── In Arazona: {}", format!("{}", datetime.with_timezone(&chrono_tz::US::Arizona)).bold());
-        println!("    └── In Pacific: {}", format!("{}", datetime.with_timezone(&chrono_tz::US::Pacific)).bold());
+        println!(
+            "  ├── Date 'human': {}",
+            format!("{}", datetime.format("%b %e %T %Y")).bold()
+        );
+        println!(
+            "  ├── Date in M/D/Y: {}",
+            format!(
+                "{}/{}/{}",
+                datetime.month(),
+                datetime.day(),
+                datetime.year()
+            )
+            .bold()
+        );
+        println!(
+            "  ├── Date in YMD: {}",
+            format!(
+                "{}{:02}{:02}",
+                datetime.year(),
+                datetime.month(),
+                datetime.day()
+            )
+            .bold()
+        );
+        println!(
+            "  ├── Day in year: {}",
+            format!("{}", datetime.ordinal()).bold()
+        );
+        println!(
+            "  ├── ISO week {}",
+            format!(
+                "{}-{}",
+                datetime.iso_week().year(),
+                datetime.iso_week().week()
+            )
+            .bold()
+        );
+        println!(
+            "  ├── Day of week: {}",
+            format!("{:?}", datetime.weekday()).bold()
+        );
+        println!(
+            "  └── Time: {}",
+            format!("{} {}", datetime.time(), datetime.timezone()).bold()
+        );
+        println!(
+            "    ├── Unix epoch(s):  {}",
+            format!("{}", datetime.timestamp()).bold()
+        );
+        println!(
+            "    ├── Unix epoch(ms): {}",
+            format!("{}", datetime.timestamp_millis()).bold()
+        );
+        println!(
+            "    ├── Unix epoch(ns): {}",
+            format!("{}", datetime.timestamp_nanos()).bold()
+        );
+        println!(
+            "    ├── In UTC: {}",
+            format!("{}", datetime.with_timezone(&chrono::Utc)).bold()
+        );
+        println!(
+            "    ├── In Eastern: {}",
+            format!("{}", datetime.with_timezone(&chrono_tz::US::Eastern)).bold()
+        );
+        println!(
+            "    ├── In Central: {}",
+            format!("{}", datetime.with_timezone(&chrono_tz::US::Central)).bold()
+        );
+        println!(
+            "    ├── In Mountain: {}",
+            format!("{}", datetime.with_timezone(&chrono_tz::US::Mountain)).bold()
+        );
+        println!(
+            "    ├── In Arazona: {}",
+            format!("{}", datetime.with_timezone(&chrono_tz::US::Arizona)).bold()
+        );
+        println!(
+            "    └── In Pacific: {}",
+            format!("{}", datetime.with_timezone(&chrono_tz::US::Pacific)).bold()
+        );
     }
 
     Ok(())
 }
 
 fn render_utc(input: TimeResult) -> Result<(), CliError> {
     let datetime = match input {
-        TimeResult::Epoch(epoch) => {
-            epoch.to_utc_date_time()
-        },
-        TimeResult::String(string_format) => {
-            string_format.to_utc_date_time()
-        }
+        TimeResult::Epoch(epoch) => epoch.to_utc_date_time(),
+        TimeResult::String(string_format) => string_format.to_utc_date_time(),
     };
 
     println!("{}", datetime);

src/commands/nsq/stats.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "1024"]
 #![feature(slice_concat_ext)]
 #![feature(vec_remove_item)]
 
@@ -25,15 +26,19 @@ extern crate prettytable;
 extern crate termion;
 #[macro_use]
 extern crate lazy_static;
+extern crate atty;
+extern crate pest;
+#[macro_use]
+extern crate pest_derive;
 
 mod commands;
 
 use clap::App;
 
 use commands::har::exec::do_har_command;
-use commands::json::do_json_latest_command;
+use commands::json::*;
 use commands::nsq::post::do_send_command;
-use commands::nsq::stats::do_status_command;
+use commands::nsq::stats::do_stats_command;
 use commands::time::do_time_command;
 use kopy_common_lib::configure_logging;
 
@@ -54,11 +59,12 @@ fn main() {
         ("har", Some(har_matches)) => do_har_command(har_matches),
         ("json", Some(json_matches)) => match json_matches.subcommand() {
             ("latest", Some(filter_matches)) => do_json_latest_command(filter_matches),
+            ("sql", Some(filter_matches)) => do_json_sql_command(filter_matches),
             _ => unreachable!(),
         },
         ("nsq", Some(nsq_matches)) => match nsq_matches.subcommand() {
             ("send", Some(send_matches)) => do_send_command(send_matches),
-            ("status", Some(send_matches)) => do_status_command(send_matches),
+            ("stats", Some(send_matches)) => do_stats_command(send_matches),
             _ => unreachable!(),
         },
         _ => unreachable!(),