Merge branch 'metadata-and-discovery' into read-events

Author: Snapstromegon

Makefile

@@ -2,8 +2,12 @@ qa: analyze test
 
 analyze:
 	@cargo clippy
+	@cargo fmt --check
 
 test:
 	@cargo test
 
-.PHONY: analyze qa test
+format:
+	@cargo fmt
+
+.PHONY: analyze format qa test

src/client.rs

@@ -40,7 +40,7 @@ use url::Url;
 pub struct Client {
     base_url: Url,
     api_token: String,
-    client: reqwest::Client,
+    reqwest: reqwest::Client,
 }
 
 impl Client {
@@ -49,7 +49,7 @@ impl Client {
         Client {
             base_url,
             api_token: api_token.into(),
-            client: reqwest::Client::new(),
+            reqwest: reqwest::Client::new(),
         }
     }
 
@@ -95,8 +95,8 @@ impl Client {
             .map_err(ClientError::URLParseError)?;
 
         let request = match endpoint.method() {
-            reqwest::Method::GET => self.client.get(url),
-            reqwest::Method::POST => self.client.post(url),
+            reqwest::Method::GET => self.reqwest.get(url),
+            reqwest::Method::POST => self.reqwest.post(url),
             _ => return Err(ClientError::InvalidRequestMethod),
         }
         .header("Authorization", format!("Bearer {}", self.api_token));
@@ -345,7 +345,7 @@ impl Client {
     /// }
     /// # })
     /// ```
-    /// 
+    ///
     /// # Errors
     /// This function will return an error if the request fails or if the URL is invalid.
     pub async fn list_event_types(

src/client/client_request.rs

@@ -77,12 +77,9 @@ pub trait StreamingRequest: ClientRequest {
     fn lines_stream(
         response: reqwest::Response,
     ) -> impl Stream<Item = Result<String, ClientError>> {
-        let bytes = response.bytes_stream().map_err(|err| {
-            io::Error::new(
-                io::ErrorKind::Other,
-                format!("Failed to read response stream: {err}"),
-            )
-        });
+        let bytes = response
+            .bytes_stream()
+            .map_err(|err| io::Error::other(format!("Failed to read response stream: {err}")));
         let stream_reader = StreamReader::new(bytes);
         LinesStream::new(BufReader::new(stream_reader).lines()).map_err(ClientError::from)
     }

src/client/client_request/list_subjects.rs

@@ -12,15 +12,15 @@ pub struct ListSubjectsRequest<'a> {
     pub base_subject: &'a str,
 }
 
-impl<'a> ClientRequest for ListSubjectsRequest<'a> {
+impl ClientRequest for ListSubjectsRequest<'_> {
     const URL_PATH: &'static str = "/api/v1/read-subjects";
     const METHOD: Method = Method::POST;
 
     fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
         Some(Ok(self))
     }
 }
-impl<'a> StreamingRequest for ListSubjectsRequest<'a> {
+impl StreamingRequest for ListSubjectsRequest<'_> {
     type ItemType = String;
 
     fn build_stream(

src/client/client_request/register_event_schema.rs

@@ -23,15 +23,15 @@ impl<'a> RegisterEventSchemaRequest<'a> {
     }
 }
 
-impl<'a> ClientRequest for RegisterEventSchemaRequest<'a> {
+impl ClientRequest for RegisterEventSchemaRequest<'_> {
     const URL_PATH: &'static str = "/api/v1/register-event-schema";
     const METHOD: Method = Method::POST;
 
     fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
         Some(Ok(self))
     }
 }
-impl<'a> OneShotRequest for RegisterEventSchemaRequest<'a> {
+impl OneShotRequest for RegisterEventSchemaRequest<'_> {
     type Response = ManagementEvent;
 
     fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {

src/container.rs

@@ -114,7 +114,7 @@ impl ContainerBuilder {
         Ok(Container {
             internal_port: self.internal_port,
             api_token: self.api_token.clone(),
-            container: GenericImage::new(self.image_name, self.image_tag)
+            instance: GenericImage::new(self.image_name, self.image_tag)
                 .with_exposed_port(self.internal_port)
                 .with_wait_for(WaitFor::Http(Box::new(
                     HttpWaitStrategy::new("/api/v1/ping")
@@ -137,10 +137,10 @@ impl ContainerBuilder {
 }
 
 /// A running test container for the [EventSourcingDB](https://www.eventsourcingdb.io/).
-/// 
+///
 /// Aside from managing the container, this struct also provides methods to get the data needed to connect to
 /// the database or even a fully configured client.
-/// 
+///
 /// You'll most likely want to use the [`Container::start_default`] method to create a new container instance for your tests.
 /// For more details, see the [`crate::container`] module documentation.
 /// ```
@@ -152,7 +152,7 @@ impl ContainerBuilder {
 /// ```
 #[derive(Debug)]
 pub struct Container {
-    container: ContainerAsync<GenericImage>,
+    instance: ContainerAsync<GenericImage>,
     internal_port: ContainerPort,
     api_token: String,
 }
@@ -167,7 +167,7 @@ impl Container {
     }
 
     /// Shortcut method to start the container with default settings.
-    /// 
+    ///
     /// This is the same as calling [`Container::builder`] and then [`ContainerBuilder::start`].
     /// In most cases this will create a contaienr with the latest image tag and a working configuration.
     ///
@@ -178,26 +178,23 @@ impl Container {
     }
 
     /// Get the host of the container.
-    /// 
+    ///
     /// This is the host that you can use to connect to the database. In most cases this will be `localhost`.
     ///
     /// # Errors
     /// This function will return an error if the container is not running (e.g. because it crashed) or if the host could not be retrieved
     pub async fn get_host(&self) -> Result<Host, ContainerError> {
-        Ok(self.container.get_host().await?)
+        Ok(self.instance.get_host().await?)
     }
 
     /// Get the mapped port for the database.
-    /// 
+    ///
     /// This is the port that you can use to connect to the database. This will be a random port that is mapped to the internal port configured via [`ContainerBuilder::with_port`].
     ///
     /// # Errors
     /// This function will return an error if the container is not running (e.g. because it crashed) or if the host could not be retrieved
     pub async fn get_mapped_port(&self) -> Result<u16, ContainerError> {
-        Ok(self
-            .container
-            .get_host_port_ipv4(self.internal_port)
-            .await?)
+        Ok(self.instance.get_host_port_ipv4(self.internal_port).await?)
     }
 
     /// Get the complete http base URL for the database.
@@ -227,7 +224,7 @@ impl Container {
     /// # Errors
     /// This function will return an error if the container could not be stopped.
     pub async fn stop(self) -> Result<(), ContainerError> {
-        self.container.stop().await?;
+        self.instance.stop().await?;
         Ok(())
     }
 

src/event.rs

@@ -10,5 +10,5 @@ pub use event_types::event_candidate::EventCandidate;
 pub use event_types::management_event::ManagementEvent;
 pub use trace_info::TraceInfo;
 
-#[cfg(feature="cloudevents")]
+#[cfg(feature = "cloudevents")]
 use crate::error::EventError;

src/event/event_types/event.rs

@@ -2,8 +2,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
-use crate::event::{trace_info::TraceInfo, EventCandidate};
-
+use crate::event::{EventCandidate, trace_info::TraceInfo};
 
 /// Represents an event that has been received from the DB.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

src/event/event_types/event_candidate.rs

@@ -1,7 +1,7 @@
+use crate::event::trace_info::TraceInfo;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use typed_builder::TypedBuilder;
-use crate::event::trace_info::TraceInfo;
 
 #[cfg(feature = "cloudevents")]
 use crate::error::EventError;

src/event/event_types/management_event.rs

@@ -1,11 +1,11 @@
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use chrono::{DateTime, Utc};
 
 /// Represents a management event that has been received from the DB.
-/// 
+///
 /// For management requests like [`crate::client::Client::ping`] and [`crate::client::Client::verify_api_token`] the DB will send a management event.
-/// 
+///
 /// Compared to a normal Event, this does not contain the following fields:
 /// - hash
 /// - predecessorhash
@@ -18,7 +18,7 @@ pub struct ManagementEvent {
     source: String,
     specversion: String,
     subject: String,
-    time: DateTime<Utc>, 
+    time: DateTime<Utc>,
     r#type: String,
 }
 
@@ -61,7 +61,7 @@ impl ManagementEvent {
         &self.time
     }
     /// Get the type of an event.
-    /// 
+    ///
     /// This method is called `ty` to avoid conflicts with the `type` keyword in Rust.
     #[must_use]
     pub fn ty(&self) -> &str {