Support runtime Topic designation

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hedwig"
-version = "6.3.0"
+version = "6.4.0"
 authors = [
     "Aniruddha Maru <[email protected]>",
     "Simonas Kazlauskas <[email protected]>",
@@ -39,6 +39,7 @@ bytes = "1"
 either = { version = "1", features = ["use_std"], default-features = false }
 futures-util = { version = "0.3.17", features = ["std", "sink"], default-features = false }
 pin-project = "1"
+smallstr = { version = "0.3.0", features = ["union"] }
 thiserror = { version = "1", default-features = false }
 url = { version = "2", default-features = false }
 uuid = { version = "^0.8", features = ["v4"], default-features = false }

src/backends/googlepubsub/publisher.rs

@@ -374,16 +374,21 @@ where
                 let sink = {
                     let retry_policy = this.retry_policy;
                     let response_sink = this.response_sink;
-                    this.topic_sinks.entry(topic.clone()).or_insert_with(|| {
-                        Box::pin(TopicSink::new(
-                            client.client.publish_topic_sink(
-                                TopicName::new(topic.as_ref())
-                                    .into_project_topic_name(client.project()),
+
+                    // avoid cloning the topic if the key exists
+                    match this.topic_sinks.get_mut(&topic) {
+                        Some(existing) => existing,
+                        None => this.topic_sinks.entry(topic.clone()).or_insert(Box::pin(
+                            TopicSink::new(
+                                client.client.publish_topic_sink(
+                                    TopicName::new(topic.as_ref())
+                                        .into_project_topic_name(client.project()),
+                                ),
+                                retry_policy.clone(),
+                                Shared::clone(response_sink),
                             ),
-                            retry_policy.clone(),
-                            Shared::clone(response_sink),
-                        ))
-                    })
+                        )),
+                    }
                 };
 
                 // poll the sink to see if it's ready

src/topic.rs

@@ -1,21 +1,36 @@
+use smallstr::SmallString;
+
 /// A message queue topic name to which messages can be published
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
-pub struct Topic(&'static str);
+// A survey of common topics found lengths between 16 and 35 bytes
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct Topic(SmallString<[u8; 36]>);
+
+impl Default for Topic {
+    fn default() -> Self {
+        Topic(SmallString::new())
+    }
+}
 
 impl std::fmt::Display for Topic {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        std::fmt::Display::fmt(self.0, f)
+        std::fmt::Display::fmt(self.0.as_str(), f)
+    }
+}
+
+impl<'a> From<&'a str> for Topic {
+    fn from(s: &'a str) -> Topic {
+        Topic(s.into())
     }
 }
 
-impl From<&'static str> for Topic {
-    fn from(s: &'static str) -> Topic {
-        Topic(s)
+impl From<String> for Topic {
+    fn from(s: String) -> Topic {
+        Topic(s.into())
     }
 }
 
 impl AsRef<str> for Topic {
     fn as_ref(&self) -> &str {
-        self.0
+        self.0.as_ref()
     }
 }