Fix seek-rewind when using nanosecond timestamps

sea-streamer-redis/src/consumer/mod.rs

@@ -60,6 +60,7 @@ struct ConsumerConfig {
     consumer_id: Option<ConsumerId>,
     auto_ack: bool,
     pre_fetch: bool,
+    timestamp_format: TimestampFormat,
 }
 
 /// More constants used throughout SeaStreamer Redis.
@@ -86,8 +87,15 @@ impl Consumer for RedisConsumer {
 
     #[inline]
     async fn seek(&mut self, ts: Timestamp) -> RedisResult<()> {
-        self.seek_to(((ts.unix_timestamp_nanos() / 1_000_000) as u64, u16::MAX))
-            .await
+        self.seek_to((
+            (match self.config.timestamp_format {
+                TimestampFormat::UnixTimestampMillis => ts.unix_timestamp_nanos() / 1_000_000,
+                #[cfg(feature = "nanosecond-timestamp")]
+                TimestampFormat::UnixTimestampNanos => ts.unix_timestamp_nanos(),
+            }) as u64,
+            u16::MAX,
+        ))
+        .await
     }
 
     #[inline]

sea-streamer-redis/src/consumer/options.rs

@@ -57,6 +57,7 @@ impl From<&RedisConsumerOptions> for ConsumerConfig {
             consumer_id: options.consumer_id().cloned(),
             auto_ack: options.auto_commit() == &AutoCommit::Delayed,
             pre_fetch: options.pre_fetch(),
+            timestamp_format: options.timestamp_format,
         }
     }
 }

sea-streamer-redis/tests/seek-rewind.rs

@@ -30,7 +30,9 @@ async fn main() -> anyhow::Result<()> {
     async fn test(consumer_mode: ConsumerMode, batch_size: usize) -> anyhow::Result<()> {
         println!("ConsumerMode = {consumer_mode:?}; batch_size = {batch_size} ...");
 
-        let options = RedisConnectOptions::default();
+        let mut options = RedisConnectOptions::default();
+        #[cfg(feature = "nanosecond-timestamp")]
+        options.set_timestamp_format(sea_streamer_redis::TimestampFormat::UnixTimestampNanos);
         let streamer = RedisStreamer::connect(
             std::env::var("BROKERS_URL")
                 .unwrap_or_else(|_| "redis://localhost".to_owned())