Version 0.2

Cargo.toml

@@ -1,19 +1,44 @@
 [package]
-name = "changes-stream"
+name = "changes-stream2"
 description = "couchdb follower"
-version = "0.1.0"
-authors = ["Ashley Williams <[email protected]>"]
+version = "0.2.19"
+authors = [
+    "Ashley Williams <[email protected]>",
+    "René Rössler <[email protected]>",
+]
+repository = "https://github.com/elwerene/changes-stream-rust.git"
 license = "MIT"
+edition = "2021"
 
 [dependencies]
-futures = "0.1.10"
-tokio-core = "0.1.4"
-serde = "0.8"
-serde_json = "0.8"
-serde_derive = "0.8"
+bytes = "1"
+futures-util = "0.3"
+serde_json = "1"
+log = "0.4"
+metrics = { version = "0.24", optional = true }
+regex = { version = "1", optional = true }
+reqwest = { version = "0.12", default-features = false, features = [
+    "stream",
+    "json",
+] }
+serde = "1"
+serde_derive = "1"
+thiserror = "2"
+tokio = "1"
+url = "2"
 
-[dependencies.hyper]
-git = "https://github.com/hyperium/hyper.git"
+[dev-dependencies]
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 
-[dependencies.hyper-tls]
-git = "https://github.com/hyperium/hyper-tls.git"
+[features]
+default = ["native-tls"]
+
+# change.doc as serde_json::value::RawValue
+raw_value_doc = ["serde_json/raw_value"]
+
+# metrics
+metrics = ["dep:metrics", "dep:regex"]
+
+# tls library selection
+native-tls = ["reqwest/native-tls"]
+rustls-tls = ["reqwest/rustls-tls"]

README.md

@@ -1,49 +1,64 @@
-# changes-stream-rust
+# changes_stream2
 
-[![travis badge](https://travis-ci.org/ashleygwilliams/changes-stream-rust.svg?branch=master)](https://travis-ci.org/ashleygwilliams/changes-stream-rust)
+[![License](https://img.shields.io/badge/license-MIT-blue.svg)](
+https://github.com/elwerene/changes-stream-rust/blob/master/LICENSE)
+[![Cargo](https://img.shields.io/crates/v/changes-stream2.svg)](
+https://crates.io/crates/changes-stream2)
+[![Documentation](https://docs.rs/changes-stream2/badge.svg)](
+https://docs.rs/changes-stream2)
 
+Fork of https://github.com/ashleygwilliams/changes-stream-rust / https://crates.io/crates/changes-stream.
 
-an implementation of [`changes-stream`](https://github.com/jcrugzz/changes-stream) in Rust.
+An implementation of [`changes-stream`](https://github.com/jcrugzz/changes-stream) in Rust.
 
-this code reads in a readable stream from an endpoint and returns each chunk in JSON.
+This code reads in a readable stream from an endpoint, parses each line and returns CouchDB changes events as defined in [src/event.rs](/src/event.rs).
 
-this code works off of the [`tokio` branch] of [`hyper`] to take advantage of new Rust Futures.
-
-[`tokio` branch]: https://github.com/hyperium/hyper/tree/tokio
-[`hyper`]: https:///crates.io/crates/hyper
-
-`changes-stream-rust` only works on nightly because it uses [`serde`].
-
-[`serde`]: https://crates.io/crates/serde
 
 ## usage
 
 in your `Cargo.toml`:
 
 ```toml
 [dependencies]
-changes-stream = { git = "https://github.com/ashleygwilliams/changes-stream-rust.git" }
+changes-stream2 = "0.2"
 ```
 
 from [examples/follower.rs](/examples/follower.rs):
 
 ```rust
-extern crate changes_stream;
-extern crate futures;
+use changes_stream2::{ChangesStream, Event};
+use futures_util::stream::StreamExt;
 
-use std::io;
-use std::io::Write;
+#[tokio::main]
+async fn main() {
+    let url = "https://replicate.npmjs.com/_changes".to_string();
+    let mut changes = ChangesStream::new(url).await.unwrap();
+    while let Some(event) = changes.next().await {
+        match event {
+            Ok(Event::Change(change)) => println!("Change ({}): {}", change.seq, change.id),
+            Ok(Event::Finished(finished)) => println!("Finished: {}", finished.last_seq),
+            Err(err) => println!("Error: {:?}", err),
+        }
+    }
+}
+```
 
-use changes_stream::ChangesStream;
+## features
 
-fn main() {
-    let url = "https://replicate.npmjs.com/_changes".to_string();
-    let mut changes = ChangesStream::new(url);
+### metrics
 
-    changes.on(|change| {
-        io::stdout().write_all(&change).unwrap();
-    });
+Enables metrics collection of the changes stream as counter values. The name is generated from the host and path of the url(database name). The metrics are:
+ * `couchdb_changes_{name}_bytes`: Total bytes read from the changes stream
+ * `couchdb_changes_{name}_entries`: Total parsed change entries
 
-    changes.run();
-}
-```
+### raw_value_doc
+
+Changes the type of ChangeEvent::Doc from `serde_json::Map<String, serde_json::Value>` to `serde_json::value::RawValue`.
+
+### native-tls
+
+Use the native-tls crate for TLS connections. This is the default.
+
+### rustls-tls
+
+Use the rustls crate for TLS connections.

examples/follower.rs

@@ -1,14 +1,15 @@
-extern crate serde_json;
-extern crate changes_stream;
-extern crate futures;
+use changes_stream2::{ChangesStream, Event};
+use futures_util::stream::StreamExt;
 
-use changes_stream::ChangesStream;
-
-fn main() {
-    let url = String::from("https://replicate.npmjs.com/_changes");
-    let mut changes = ChangesStream::new(url);
-    changes.on(|change| {
-        println!("{}: {}", change.seq, change.id);
-    });
-    changes.run();
+#[tokio::main]
+async fn main() {
+    let url = "https://replicate.npmjs.com/_changes".to_string();
+    let mut changes = ChangesStream::new(url).await.unwrap();
+    while let Some(event) = changes.next().await {
+        match event {
+            Ok(Event::Change(change)) => println!("Change ({}): {}", change.seq, change.id),
+            Ok(Event::Finished(finished)) => println!("Finished: {}", finished.last_seq),
+            Err(err) => println!("Error: {:?}", err),
+        }
+    }
 }

src/error.rs

@@ -0,0 +1,20 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Could not parse the url")]
+    InvalidUrl(#[from] url::ParseError),
+    #[error("Request failed")]
+    RequestFailed(#[from] reqwest::Error),
+    #[error("Server answered with non-ok status: {status}. body: {body}")]
+    InvalidResponse {
+        status: reqwest::StatusCode,
+        body: String,
+    },
+    #[error("Could not parse server response: {json}")]
+    ParsingFailed {
+        #[source]
+        error: serde_json::Error,
+        json: String,
+    },
+}

src/event.rs

@@ -1,11 +1,35 @@
+use serde_derive::{Deserialize, Serialize};
+use serde_json::Value;
+
 #[derive(Serialize, Deserialize, Debug)]
-pub struct Event {
-    pub seq: u64,
+pub enum Event {
+    Change(ChangeEvent),
+    Finished(FinishedEvent),
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ChangeEvent {
+    pub seq: Value,
     pub id: String,
-    pub changes: Vec<Change>
+    pub changes: Vec<Change>,
+
+    #[serde(default)]
+    pub deleted: bool,
+
+    #[serde(default)]
+    #[cfg(feature = "raw_value_doc")]
+    pub doc: Option<Box<serde_json::value::RawValue>>,
+    #[cfg(not(feature = "raw_value_doc"))]
+    pub doc: Option<serde_json::Map<String, Value>>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
+pub struct FinishedEvent {
+    pub last_seq: Value,
+    pub pending: Option<u64>, // not available on CouchDB 1.0
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Change {
     pub rev: String,
 }