Make Topic a newtype (#29)

Turns out having anything with any sort of lifetime in stream/sink
chains will trigger nasty issues such as
rust-lang/rust#79648 which is pretty difficult
to figure out how to work around.

Making `Topic` a newtype erases the lifetime from the type, making it
significantly easier to work with in thsoe contexts.

Author: nagisa

examples/publish.rs

@@ -15,8 +15,8 @@ struct UserCreatedMessage {
 impl<'a> EncodableMessage for &'a UserCreatedMessage {
     type Error = hedwig::validators::JsonSchemaValidatorError;
     type Validator = hedwig::validators::JsonSchemaValidator;
-    fn topic(&self) -> &'static str {
-        "user.created"
+    fn topic(&self) -> hedwig::Topic {
+        "user.created".into()
     }
     fn encode(self, validator: &Self::Validator) -> Result<hedwig::ValidatedMessage, Self::Error> {
         Ok(validator

src/lib.rs

@@ -58,7 +58,7 @@
 //! impl<'a> hedwig::publish::EncodableMessage for &'a UserCreatedMessage {
 //!     type Error = hedwig::validators::JsonSchemaValidatorError;
 //!     type Validator = hedwig::validators::JsonSchemaValidator;
-//!     fn topic(&self) -> hedwig::Topic { "user.created" }
+//!     fn topic(&self) -> hedwig::Topic { "user.created".into() }
 //!     fn encode(self, validator: &Self::Validator)
 //!     -> Result<hedwig::ValidatedMessage, Self::Error> {
 //!         validator.validate(
@@ -106,20 +106,17 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
 use std::{collections::BTreeMap, time::SystemTime};
-
+pub use topic::Topic;
 use uuid::Uuid;
 
 #[cfg(feature = "publish")]
 #[cfg_attr(docsrs, doc(cfg(feature = "publish")))]
 pub mod publish;
-
 #[cfg(test)]
 mod tests;
+mod topic;
 pub mod validators;
 
-/// A message queue topic name to which messages can be published
-pub type Topic = &'static str;
-
 /// All errors that may be returned when operating top level APIs.
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]

src/publish/mod.rs

@@ -185,7 +185,7 @@ where
             None => None,
             Some(stream_item) => Some((
                 stream_item,
-                this.topic,
+                *this.topic,
                 this.messages
                     .next()
                     .expect("should be as many messages as publishes"),

src/publish/publishers/mock.rs

@@ -45,7 +45,8 @@ impl MockPublisher {
     /// be published, was indeed published
     ///
     /// Panics if the message was not published.
-    pub fn assert_message_published(&self, topic: Topic, uuid: &Uuid) {
+    pub fn assert_message_published<T: Into<Topic>>(&self, topic: T, uuid: &Uuid) {
+        let topic = topic.into();
         {
             let lock = self.0.lock().expect("this mutex cannot get poisoned");
             for (mt, msg) in &lock[..] {

src/publish/publishers/null.rs

@@ -20,7 +20,7 @@ impl Publisher for NullPublisher {
     type MessageError = std::convert::Infallible;
     type PublishStream = NullPublishStream;
 
-    fn publish<'a, I>(&self, _: &'static str, messages: I) -> Self::PublishStream
+    fn publish<'a, I>(&self, _: crate::Topic, messages: I) -> Self::PublishStream
     where
         I: Iterator<Item = &'a ValidatedMessage> + ExactSizeIterator,
     {

src/publish/sink.rs

@@ -272,7 +272,7 @@ mod test {
         type Validator = TestValidator;
 
         fn topic(&self) -> Topic {
-            "test_topic"
+            "test_topic".into()
         }
 
         fn encode(self, _: &Self::Validator) -> Result<ValidatedMessage, Self::Error> {
@@ -382,7 +382,7 @@ mod test {
             assert_eq!(1, sink.sink.poll_close_called);
             assert_eq!(
                 vec![(
-                    "test_topic",
+                    "test_topic".into(),
                     TestMessage("foo").encode(&TestValidator).unwrap()
                 )],
                 sink.sink.elements
@@ -517,14 +517,14 @@ mod test {
             assert_eq!(
                 Ok(()),
                 sink.as_mut()
-                    .start_send(("test_topic", test_validated_message("foo")))
+                    .start_send(("test_topic".into(), test_validated_message("foo")))
             );
         }
 
         /// The publisher should start flushing when the batch size has been exceeded
         #[test]
         fn batching_batches() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 3;
             let publisher = MockPublisher::new();
             let sink = publisher_sink(publisher, batch_size);
@@ -565,7 +565,7 @@ mod test {
         /// The publisher should flush buffered elements when asked to close
         #[test]
         fn close_flushes_batch() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 3;
             let publisher = MockPublisher::new();
             let sink = publisher_sink(publisher, batch_size);
@@ -599,7 +599,7 @@ mod test {
         /// The publisher should flush buffered elements when asked to flush
         #[test]
         fn flush_incomplete_batch() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 3;
             let publisher = MockPublisher::new();
             let sink = publisher_sink(publisher, batch_size);
@@ -635,7 +635,7 @@ mod test {
         #[test]
         #[should_panic]
         fn panic_at_buffer_full_without_ready_check() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 1;
             let publisher = MockPublisher::new();
             let sink = publisher_sink(publisher, batch_size);
@@ -656,7 +656,7 @@ mod test {
         /// Step through flushing a non-full batch and see that yield points are respected
         #[test]
         fn partial_flushing_check() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 3;
             let (publisher, command) = ControlledPublisher::new();
             let sink = publisher_sink(publisher, batch_size);
@@ -724,7 +724,7 @@ mod test {
         /// A failed message can be re-sent to the sink and eventually succeed
         #[test]
         fn flushing_error_retry() {
-            let topic = "test_topic";
+            let topic = "test_topic".into();
             let batch_size = 5;
             let (publisher, command) = ControlledPublisher::new();
             let sink = publisher_sink(publisher, batch_size);

src/tests.rs

@@ -71,8 +71,8 @@ impl<'a, I: serde::Serialize> EncodableMessage for &'a JsonUserCreatedMessage<I>
     type Error = validators::JsonSchemaValidatorError;
     type Validator = validators::JsonSchemaValidator;
 
-    fn topic(&self) -> &'static str {
-        "user.created"
+    fn topic(&self) -> crate::Topic {
+        "user.created".into()
     }
     fn encode(self, validator: &Self::Validator) -> Result<ValidatedMessage, Self::Error> {
         validator.validate(

src/topic.rs

@@ -0,0 +1,21 @@
+/// A message queue topic name to which messages can be published
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
+pub struct Topic(&'static str);
+
+impl std::fmt::Display for Topic {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        std::fmt::Display::fmt(self.0, f)
+    }
+}
+
+impl From<&'static str> for Topic {
+    fn from(s: &'static str) -> Topic {
+        Topic(s)
+    }
+}
+
+impl From<Topic> for &'static str {
+    fn from(s: Topic) -> &'static str {
+        s.0
+    }
+}