fix: Ability to set deliver_at_time from SerializeMessage messages directly (streamnative#337)

streamnative#336

Before this, the only way to set deliver_at_time was via the MessageBuilder.

Co-authored-by: Rui Fu <freeznet@users.noreply.github.com>

Author: 2 people

examples/consumer.rs

@@ -50,8 +50,7 @@ async fn main() -> Result<(), pulsar::Error> {
         env::var("PULSAR_BASIC_USERNAME"),
         env::var("PULSAR_BASIC_PASSWORD"),
     ) {
-        builder =
-            builder.with_auth_provider(Box::new(BasicAuthentication::new(&username, &password)))
+        builder = builder.with_auth_provider(BasicAuthentication::new(&username, &password))
     }
 
     let pulsar: Pulsar<_> = builder.build().await?;

examples/producer.rs

@@ -52,8 +52,7 @@ async fn main() -> Result<(), pulsar::Error> {
         env::var("PULSAR_BASIC_USERNAME"),
         env::var("PULSAR_BASIC_PASSWORD"),
     ) {
-        builder =
-            builder.with_auth_provider(Box::new(BasicAuthentication::new(&username, &password)))
+        builder = builder.with_auth_provider(BasicAuthentication::new(&username, &password))
     }
 
     let pulsar: Pulsar<_> = builder.build().await?;

src/authentication/basic.rs

@@ -12,10 +12,10 @@ pub struct BasicAuthentication {
 
 impl BasicAuthentication {
     #[must_use]
-    pub fn new(username: &str, password: &str) -> Self {
-        Self {
+    pub fn new(username: &str, password: &str) -> Box<Self> {
+        Box::new(Self {
             auth_data: format!("{username}:{password}"),
-        }
+        })
     }
 }
 

src/producer.rs

@@ -81,6 +81,9 @@ pub struct Message {
     pub event_time: ::std::option::Option<u64>,
     /// current version of the schema
     pub schema_version: ::std::option::Option<Vec<u8>>,
+    /// UTC Unix timestamp in milliseconds, time at which the message should be
+    /// delivered to consumers
+    pub deliver_at_time: ::std::option::Option<i64>,
 }
 
 /// internal message type carrying options that must be defined
@@ -126,6 +129,7 @@ impl From<Message> for ProducerMessage {
             replicate_to: m.replicate_to,
             event_time: m.event_time,
             schema_version: m.schema_version,
+            deliver_at_time: m.deliver_at_time,
             ..Default::default()
         }
     }