refactor: port async_json_stream_reader implementation

Author: ngutman

Cargo.toml

@@ -3,26 +3,25 @@ name = "asyncjsonparser"
 version = "0.1.0"
 edition = "2021"
 rust-version = "1.74"
-description = "Async JSON value streaming parser for AsyncRead sources"
+description = "Async JSON stream reader for selective parsing of large payloads"
 readme = "README.md"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/singular-labs/async-json-streamer"
 homepage = "https://github.com/singular-labs/async-json-streamer"
 documentation = "https://docs.rs/asyncjsonparser"
-keywords = ["json", "async", "streaming", "parser", "ndjson"]
+keywords = ["json", "async", "streaming", "parser"]
 categories = ["asynchronous", "parser-implementations", "encoding"]
 
 [dependencies]
-bytes = "1.6"
-futures = "0.3"
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
-thiserror = "1.0"
+bytes = "1.4.0"
+serde_json = "1.0.96"
+thiserror = "2.0.11"
+tokio = { version = "1.34.0", features = ["fs", "io-util", "macros", "rt-multi-thread"] }
+tracing = "0.1.40"
 
 [dev-dependencies]
-futures = { version = "0.3", features = ["executor"] }
-tokio = { version = "1.37", features = ["fs", "macros", "rt-multi-thread", "io-util"] }
-tokio-util = { version = "0.7", features = ["compat"] }
+assert_matches = "1.5.0"
+serde = { version = "1.0.199", features = ["derive"] }
 
 [package.metadata.docs.rs]
 all-features = true

README.md

@@ -1,6 +1,7 @@
 # asyncjsonparser
 
-Async JSON value streaming parser for `AsyncRead` sources.
+Async JSON stream reader for selective parsing of large payloads. This is the
+standalone home of Extract's `AsyncJsonStreamReader` implementation.
 
 [![crates.io](https://img.shields.io/crates/v/asyncjsonparser.svg)](https://crates.io/crates/asyncjsonparser)
 [![docs.rs](https://docs.rs/asyncjsonparser/badge.svg)](https://docs.rs/asyncjsonparser)
@@ -9,11 +10,11 @@ Async JSON value streaming parser for `AsyncRead` sources.
 
 ## Why asyncjsonparser
 
-- Parses **consecutive JSON values** from a single async byte stream.
-- Handles **NDJSON** and whitespace-separated JSON values.
-- Works with **any runtime** via `futures::io::AsyncRead`.
-- **Configurable buffering** with a hard limit for safety.
-- **No unsafe** code.
+- Stream through large JSON without deserializing the full payload.
+- Selectively read keys, values, and arrays using a tokenized reader.
+- Handles chunk boundaries and escaped strings correctly.
+- Built on Tokio `AsyncRead`.
+- No unsafe code.
 
 ## Install
 
@@ -25,98 +26,48 @@ cargo add asyncjsonparser
 
 ```rust
 use asyncjsonparser::AsyncJsonStreamReader;
-use futures::executor::block_on;
-use futures::io::Cursor;
-
-let data = b"{\"id\":1}\n{\"id\":2} {\"id\":3}";
-let mut parser = AsyncJsonStreamReader::new(Cursor::new(data));
-
-let ids = block_on(async {
-    let mut out = Vec::new();
-    while let Some(value) = parser.next_value::<serde_json::Value>().await? {
-        out.push(value["id"].as_i64().unwrap());
-    }
-    Ok::<_, asyncjsonparser::Error>(out)
-})
-.unwrap();
-
-assert_eq!(ids, vec![1, 2, 3]);
-```
-
-## Tokio example
-
-`asyncjsonparser` uses `futures::io::AsyncRead`. For Tokio, adapt with
-`tokio-util`:
-
-```bash
-cargo add tokio --features fs,io-util,macros,rt-multi-thread
-cargo add tokio-util --features compat
-```
-
-```rust
-use asyncjsonparser::AsyncJsonStreamReader;
-use tokio_util::compat::TokioAsyncReadCompatExt;
+use std::io::Cursor;
 
 #[tokio::main]
-async fn main() -> Result<(), asyncjsonparser::Error> {
-    let file = tokio::fs::File::open("data.ndjson").await?;
-    let reader = tokio::io::BufReader::new(file).compat();
-    let mut parser = AsyncJsonStreamReader::new(reader);
-
-    while let Some(value) = parser.next_value::<serde_json::Value>().await? {
-        println!("id={}", value["id"]);
+async fn main() -> Result<(), asyncjsonparser::AsyncJsonStreamReaderError> {
+    let data = r#"{"status":"success","results":[{"id":1},{"id":2}]}"#;
+    let mut reader = AsyncJsonStreamReader::new(Cursor::new(data.as_bytes().to_vec()));
+
+    while let Some(key) = reader.next_object_entry().await? {
+        match key.as_str() {
+            "status" => {
+                let status = reader.read_string().await?;
+                println!("status={status}");
+            }
+            "results" => {
+                while reader.start_array_item().await? {
+                    let obj = reader.deserialize_object().await?;
+                    println!("id={}", obj["id"]);
+                }
+            }
+            _ => {}
+        }
     }
 
     Ok(())
 }
 ```
 
-## Stream interface
-
-```rust
-use asyncjsonparser::AsyncJsonStreamReader;
-use futures::executor::block_on;
-use futures::io::Cursor;
-use futures::stream::TryStreamExt;
-
-let data = b"{\"id\":1} {\"id\":2}";
-let parser = AsyncJsonStreamReader::new(Cursor::new(data));
-
-let ids: Vec<i64> = block_on(async {
-    parser
-        .into_stream::<serde_json::Value>()
-        .map_ok(|value| value["id"].as_i64().unwrap())
-        .try_collect()
-        .await
-})
-.unwrap();
-
-assert_eq!(ids, vec![1, 2]);
-```
-
-## Configuration
-
-```rust
-use asyncjsonparser::{AsyncJsonStreamReader, ParserConfig};
-use futures::io::Cursor;
+## Common patterns
 
-let config = ParserConfig::new(2 * 1024 * 1024, 16 * 1024);
-let parser = AsyncJsonStreamReader::with_config(Cursor::new(b"{}"), config);
-```
+- Read object entries with `next_object_entry`.
+- Skip values by calling `next_object_entry` again without consuming the value.
+- Stream arrays with `start_array_item`.
+- Parse string/number/bool with `read_string`, `read_number`, `read_boolean`.
+- Deserialize a sub-object with `deserialize_object`.
 
 ## Error handling
 
-All fallible operations return `asyncjsonparser::Error`:
-
-- `Error::Io` for reader failures
-- `Error::Json` for invalid or truncated JSON
-- `Error::BufferLimit` when the buffer exceeds `max_buffer_size`
-
-## Performance tips
+All fallible operations return `AsyncJsonStreamReaderError`:
 
-- Increase `read_chunk_size` if your reader prefers larger reads.
-- Keep `max_buffer_size` large enough for your largest JSON payload.
-- For huge payloads, prefer NDJSON over a single giant JSON array.
+- `Io` for reader failures
+- `JsonError` for malformed JSON
+- `UnexpectedToken` when the stream doesn't match the expected structure
 
 ## MSRV
 

examples/ndjson.rs

@@ -1,15 +1,26 @@
 use asyncjsonparser::AsyncJsonStreamReader;
-use futures::executor::block_on;
-use futures::io::Cursor;
+use std::io::Cursor;
 
-fn main() -> Result<(), asyncjsonparser::Error> {
-    let data = b"{\"id\":1}\n{\"id\":2} {\"id\":3}";
-    let mut parser = AsyncJsonStreamReader::new(Cursor::new(data));
+#[tokio::main]
+async fn main() -> Result<(), asyncjsonparser::AsyncJsonStreamReaderError> {
+    let data = r#"{"status":"ok","results":[{"id":1},{"id":2}]}"#;
+    let mut reader = AsyncJsonStreamReader::new(Cursor::new(data.as_bytes().to_vec()));
 
-    block_on(async {
-        while let Some(value) = parser.next_value::<serde_json::Value>().await? {
-            println!("id={}", value["id"]);
+    while let Some(key) = reader.next_object_entry().await? {
+        match key.as_str() {
+            "status" => {
+                let status = reader.read_string().await?;
+                println!("status={status}");
+            }
+            "results" => {
+                while reader.start_array_item().await? {
+                    let obj = reader.deserialize_object().await?;
+                    println!("id={}", obj["id"]);
+                }
+            }
+            _ => {}
         }
-        Ok(())
-    })
+    }
+
+    Ok(())
 }

examples/tokio.rs

@@ -1,14 +1,17 @@
 use asyncjsonparser::AsyncJsonStreamReader;
-use tokio_util::compat::TokioAsyncReadCompatExt;
+use tokio::io::BufReader;
 
 #[tokio::main]
-async fn main() -> Result<(), asyncjsonparser::Error> {
-    let file = tokio::fs::File::open("data.ndjson").await?;
-    let reader = tokio::io::BufReader::new(file).compat();
+async fn main() -> Result<(), asyncjsonparser::AsyncJsonStreamReaderError> {
+    let file = tokio::fs::File::open("data.json").await?;
+    let reader = BufReader::new(file);
     let mut parser = AsyncJsonStreamReader::new(reader);
 
-    while let Some(value) = parser.next_value::<serde_json::Value>().await? {
-        println!("id={}", value["id"]);
+    while let Some(key) = parser.next_object_entry().await? {
+        if key == "payload" {
+            let payload = parser.deserialize_object().await?;
+            println!("payload keys: {}", payload.len());
+        }
     }
 
     Ok(())