Merge branch 'main' into fix-building-all-features

Author: dotKuro

Cargo.toml

@@ -2,6 +2,11 @@
 name = "eventsourcingdb-client-rust"
 version = "0.1.0"
 edition = "2024"
+description = "A client library for the EventsourcingDB by the native web."
+repository = "https://github.com/thenativeweb/eventsourcingdb-client-rust"
+license = "MIT"
+keywords = ["eventsourcingdb", "eventsourcing", "event-sourcing", "cqrs", "thenativeweb"]
+categories = ["SDK", "database", "database client", "event sourcing", "cqrs"]
 
 [features]
 default = []

LICENSE.md

@@ -0,0 +1,9 @@
+MIT License
+
+Copyright (c) 2025 the native web GmbH
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -10,7 +10,7 @@ Based on the [compliance criteria](https://docs.eventsourcingdb.io/client-sdks/c
 - 🚀 [Essentials](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#essentials)
 - 🚀 [Writing Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#writing-events)
 - 🚀 [Reading Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#reading-events)
-- ❌ [Using EventQL](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#using-eventql)
+- 🚀 [Using EventQL](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#using-eventql)
 - 🚀 [Observing Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#observing-events)
 - 🚀 [Metadata and Discovery](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#metadata-and-discovery)
 - 🚀 [Testcontainers Support](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#testcontainers-support)

src/client.rs

@@ -100,7 +100,7 @@ impl Client {
             reqwest::Method::POST => self.reqwest.post(url),
             _ => return Err(ClientError::InvalidRequestMethod),
         }
-        .header("Authorization", format!("Bearer {}", self.api_token));
+        .bearer_auth(&self.api_token);
         let request = if let Some(body) = endpoint.body() {
             request
                 .header("Content-Type", "application/json")
@@ -224,9 +224,22 @@ impl Client {
     /// # let db_url = container.get_base_url().await.unwrap();
     /// # let api_token = container.get_api_token();
     /// let client = eventsourcingdb_client_rust::client::Client::new(db_url, api_token);
-    /// let mut event_stream = client.observe_events("/", None).await.expect("Failed to observe events");
-    /// while let Some(event) = event_stream.next().await {
-    ///     println!("Found Type {:?}", event.expect("Error while reading event"));
+    /// # client.write_events(
+    /// #   vec![
+    /// #     EventCandidate::builder()
+    /// #        .source("https://www.eventsourcingdb.io".to_string())
+    /// #        .data(json!({"value": 1}))
+    /// #        .subject("/test".to_string())
+    /// #        .r#type("io.eventsourcingdb.test".to_string())
+    /// #        .build()
+    /// #   ],
+    /// #   vec![]
+    /// # ).await.expect("Failed to write events");
+    /// let mut event_stream = client.observe_events("/test", None).await.expect("Failed to observe events");
+    /// match event_stream.next().await {
+    ///     Some(Ok(event)) => println!("Found Event {:?}", event),
+    ///     Some(Err(e)) => eprintln!("Error while reading event: {:?}", e),
+    ///     None => println!("No more events."),
     /// }
     /// # })
     /// ```

src/client/client_request.rs

@@ -17,6 +17,7 @@ pub use ping::PingRequest;
 pub use read_events::ReadEventsRequest;
 pub use register_event_schema::RegisterEventSchemaRequest;
 pub use run_eventql_query::RunEventqlQueryRequest;
+use serde_json::Value;
 pub use verify_api_token::VerifyApiTokenRequest;
 pub use write_events::WriteEventsRequest;
 
@@ -72,7 +73,7 @@ enum StreamLineItem<T> {
     Error { error: String },
     /// A heardbeat message was sent to keep the connection alive.
     /// This is only used when observing events, but it does not hurt to have it everywhere.
-    Heartbeat,
+    Heartbeat(Value),
     /// A successful response from the database
     /// Since the exact type of the payload is not known at this point, we use this as a fallback case.
     /// Every request item gets put in here and the type can be checked later on.
@@ -114,7 +115,7 @@ pub trait StreamingRequest: ClientRequest {
                             Some(Err(ClientError::DBError(error)))
                         }
                         // A heartbeat message was sent, which we ignore.
-                        Ok(StreamLineItem::Heartbeat) => None,
+                        Ok(StreamLineItem::Heartbeat(_value)) => None,
                         // A successful response was sent with the correct type.
                         Ok(StreamLineItem::Ok { payload, ty }) if ty == Self::ITEM_TYPE_NAME => {
                             Some(Ok(payload))

src/client/client_request/observe_events.rs

@@ -15,7 +15,7 @@ pub struct ObserveEventsRequest<'a> {
 }
 
 impl ClientRequest for ObserveEventsRequest<'_> {
-    const URL_PATH: &'static str = "/api/v1/read-events";
+    const URL_PATH: &'static str = "/api/v1/observe-events";
     const METHOD: Method = Method::POST;
 
     fn body(&self) -> Option<Result<impl Serialize, ClientError>> {

tests/observe_events.rs

@@ -0,0 +1,53 @@
+mod utils;
+
+use eventsourcingdb_client_rust::container::Container;
+use futures::stream::StreamExt;
+use serde_json::json;
+use utils::create_test_eventcandidate;
+
+#[tokio::test]
+async fn observe_existing_events() {
+    let container = Container::start_default().await.unwrap();
+    let client = container.get_client().await.unwrap();
+    let event_candidate = create_test_eventcandidate("/test", json!({"value": 1}));
+    let written = client
+        .write_events(vec![event_candidate.clone()], vec![])
+        .await
+        .expect("Unable to write event");
+
+    let mut events_stream = client
+        .observe_events("/test", None)
+        .await
+        .expect("Failed to request events");
+    let events = events_stream
+        .next()
+        .await
+        .expect("Failed to read events")
+        .expect("Expected an event, but got an error");
+
+    assert_eq!(vec![events], written);
+}
+
+#[tokio::test]
+async fn keep_observing_events() {
+    let container = Container::start_default().await.unwrap();
+    let client = container.get_client().await.unwrap();
+
+    let mut events_stream = client
+        .observe_events("/test", None)
+        .await
+        .expect("Failed to observe events");
+    let event_candidate = create_test_eventcandidate("/test", json!({"value": 1}));
+    let written = client
+        .write_events(vec![event_candidate.clone()], vec![])
+        .await
+        .expect("Unable to write event");
+
+    let event = events_stream
+        .next()
+        .await
+        .expect("Failed to read events")
+        .expect("Expected an event, but got an error");
+
+    assert_eq!(vec![event], written);
+}

tests/run_eventql_query.rs

@@ -0,0 +1,16 @@
+use eventsourcingdb_client_rust::container::Container;
+use futures::stream::TryStreamExt;
+
+#[tokio::test]
+async fn run_empty_query() {
+    let container = Container::start_default().await.unwrap();
+    let client = container.get_client().await.unwrap();
+    let rows = client
+        .run_eventql_query("FROM e IN events ORDER BY e.time DESC TOP 100 PROJECT INTO e")
+        .await
+        .expect("Unable to run query");
+    let rows: Result<Vec<_>, _> = rows.try_collect().await;
+    assert!(rows.is_ok(), "Failed to run query: {:?}", rows);
+    let rows = rows.expect("Failed to read rows");
+    assert_eq!(rows.len(), 0);
+}