Merge branch 'read-events' into observing-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

@@ -38,7 +38,7 @@ use url::Url;
 pub struct Client {
     base_url: Url,
     api_token: String,
-    client: reqwest::Client,
+    reqwest: reqwest::Client,
 }
 
 impl Client {
@@ -47,7 +47,7 @@ impl Client {
         Client {
             base_url,
             api_token: api_token.into(),
-            client: reqwest::Client::new(),
+            reqwest: reqwest::Client::new(),
         }
     }
 
@@ -93,8 +93,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));
@@ -376,7 +376,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

@@ -79,12 +79,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/read_events.rs

@@ -13,7 +13,7 @@ pub struct ReadEventsRequest<'a> {
     pub options: Option<ReadEventsRequestOptions<'a>>,
 }
 
-impl<'a> ClientRequest for ReadEventsRequest<'a> {
+impl ClientRequest for ReadEventsRequest<'_> {
     const URL_PATH: &'static str = "/api/v1/read-events";
     const METHOD: Method = Method::POST;
 
@@ -22,7 +22,7 @@ impl<'a> ClientRequest for ReadEventsRequest<'a> {
     }
 }
 
-impl<'a> StreamingRequest for ReadEventsRequest<'a> {
+impl StreamingRequest for ReadEventsRequest<'_> {
     type ItemType = Event;
 
     fn build_stream(
@@ -32,14 +32,14 @@ impl<'a> StreamingRequest for ReadEventsRequest<'a> {
         #[serde(tag = "type", content = "payload", rename_all = "camelCase")]
         enum LineItem {
             Error { error: String },
-            Event(Event),
+            Event(Box<Event>),
         }
 
         impl From<LineItem> for Result<Event, ClientError> {
             fn from(item: LineItem) -> Self {
                 match item {
                     LineItem::Error { error } => Err(ClientError::DBError(error)),
-                    LineItem::Event(event_type) => Ok(event_type),
+                    LineItem::Event(event_type) => Ok(*event_type),
                 }
             }
         }

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;