feat: add message identifier

sea-streamer-kafka/src/consumer.rs

@@ -8,8 +8,9 @@ use sea_streamer_runtime::spawn_blocking;
 use std::{collections::HashSet, fmt::Debug, time::Duration};
 
 use sea_streamer_types::{
-    Consumer as ConsumerTrait, ConsumerGroup, ConsumerMode, ConsumerOptions, Message, Payload,
-    SeqNo, SeqPos, ShardId, StreamErr, StreamKey, StreamerUri, Timestamp,
+    Consumer as ConsumerTrait, ConsumerGroup, ConsumerMode, ConsumerOptions, Message,
+    MessageIdentifier, Payload, SeqNo, SeqPos, ShardId, StreamErr, StreamKey, StreamerUri,
+    Timestamp,
     export::futures::{
         FutureExt, StreamExt,
         future::Map,
@@ -511,7 +512,7 @@ impl KafkaConsumer {
     /// and this Consumer will be unusable for any operations until it finishes.
     pub async fn commit_with(
         &mut self,
-        (stream_key, shard_id, sequence): &(StreamKey, ShardId, SeqNo),
+        (stream_key, shard_id, sequence): &MessageIdentifier,
     ) -> KafkaResult<()> {
         self.commit(stream_key, shard_id, sequence).await
     }
@@ -570,7 +571,7 @@ impl KafkaConsumer {
     /// You must have `set_enable_auto_offset_store` to false.
     pub fn store_offset_with(
         &mut self,
-        (stream_key, shard_id, sequence): &(StreamKey, ShardId, SeqNo),
+        (stream_key, shard_id, sequence): &MessageIdentifier,
     ) -> KafkaResult<()> {
         self.store_offset(stream_key, shard_id, sequence)
     }

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

@@ -21,7 +21,7 @@ use crate::{
 use sea_streamer_runtime::{spawn_task, timeout};
 use sea_streamer_types::{
     Buffer, ConnectOptions, Consumer, ConsumerGroup, ConsumerId, ConsumerMode, ConsumerOptions,
-    Message, MessageHeader, SEA_STREAMER_INTERNAL, SeqNo, SeqPos, ShardId, SharedMessage,
+    Message, MessageHeader, MessageIdentifier, SEA_STREAMER_INTERNAL, SeqPos, SharedMessage,
     StreamErr, StreamKey, Timestamp, export::futures::FutureExt,
 };
 
@@ -196,7 +196,7 @@ impl RedisConsumer {
 
     pub fn ack_with(
         &self,
-        (stream_key, shard_id, sequence): &(StreamKey, ShardId, SeqNo),
+        (stream_key, shard_id, sequence): &MessageIdentifier,
     ) -> RedisResult<()> {
         if self.config.auto_ack {
             return Err(StreamErr::Backend(RedisErr::InvalidClientConfig(

sea-streamer-types/src/message.rs

@@ -1,6 +1,6 @@
 use std::{str::Utf8Error, sync::Arc};
 
-use crate::{SeqNo, ShardId, StreamKey, Timestamp};
+use crate::{MessageIdentifier, SeqNo, ShardId, StreamKey, Timestamp};
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct OwnedMessage {
@@ -85,7 +85,7 @@ pub trait Message: Send {
     }
 
     /// tuple to uniquely identify a message
-    fn identifier(&self) -> (StreamKey, ShardId, SeqNo) {
+    fn identifier(&self) -> MessageIdentifier {
         (self.stream_key(), self.shard_id(), self.sequence())
     }
 }

sea-streamer-types/src/stream.rs

@@ -25,12 +25,14 @@ pub struct ShardId {
     id: u64,
 }
 
-/// The tuple (StreamKey, ShardId, SeqNo) uniquely identifies a message. Aka. offset.
 #[cfg(not(feature = "wide-seq-no"))]
 pub type SeqNo = u64;
 #[cfg(feature = "wide-seq-no")]
 pub type SeqNo = u128;
 
+/// The tuple (StreamKey, ShardId, SeqNo) uniquely identifies a message. Aka. offset.
+pub type MessageIdentifier = (StreamKey, ShardId, SeqNo);
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 /// Identifies a position in a stream.
 pub enum SeqPos {