diff --git a/sea-streamer-kafka/src/consumer.rs b/sea-streamer-kafka/src/consumer.rs index 2b090b9..07e8f16 100644 --- a/sea-streamer-kafka/src/consumer.rs +++ b/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) } diff --git a/sea-streamer-redis/src/consumer/mod.rs b/sea-streamer-redis/src/consumer/mod.rs index 93208eb..d28b7ba 100644 --- a/sea-streamer-redis/src/consumer/mod.rs +++ b/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( diff --git a/sea-streamer-types/src/message.rs b/sea-streamer-types/src/message.rs index 9badf97..01bf0f3 100644 --- a/sea-streamer-types/src/message.rs +++ b/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()) } } diff --git a/sea-streamer-types/src/stream.rs b/sea-streamer-types/src/stream.rs index 8493d15..fdd5ea6 100644 --- a/sea-streamer-types/src/stream.rs +++ b/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 {