Struct KWConsumer

pub struct KWConsumer {
    pub conf: KWConsumerConf,
    pub consumer: Arc<StreamConsumer<DefaultConsumerContext>>,
    pub admin_client: Mutex<Option<Arc<AdminClient<DefaultClientContext>>>>,
}

Fields

conf: KWConsumerConf

consumer: Arc<StreamConsumer<DefaultConsumerContext>>

admin_client: Mutex<Option<Arc<AdminClient<DefaultClientContext>>>>

Implementations

impl KWConsumer

pub fn new(conf: KWConsumerConf) -> KWResult<Self>

pub fn get_group_id(&self) -> &str

pub fn new_subscribe(conf: KWConsumerConf) -> KWResult<Self>

pub fn subscribe(&self) -> KWResult<()>

pub fn assign_split_partition_queue( &self, topic: &str, partition: i32, ) -> KWResult<Vec<StreamPartitionQueue<DefaultConsumerContext>>>

pub async fn recv<'a>(&self) -> Result<BorrowedMessage<'_>, KafkaError>

pub fn unsubscribe(&self)

pub fn assign(&self, topic_name: &str, partition: i32) -> KWResult<()>

pub fn split_partition_queue( &self, topic: &str, partition: i32, ) -> KWResult<StreamPartitionQueue<DefaultConsumerContext>>

pub fn store_offset( &self, topic: &str, partition: i32, offset: i64, ) -> KWResult<()>

pub fn fetch_group_list(&self) -> KWResult<GroupList>

Trait Implementations

impl OptionExt for KWConsumer

type AdminClient = AdminClient<DefaultClientContext>

fn get_brokers(&self) -> &str

fn get_log_level(&self) -> RDKafkaLogLevel

fn get_admin_client<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = KWResult<Arc<AdminClient<DefaultClientContext>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_new_topics(&self) -> KWResult<Vec<NewTopic<'_>>>

fn create_topic<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = KWResult<()>> + Send + 'async_trait>>

where Self: Sync + 'async_trait, 'life0: 'async_trait,

fn create_admin_client(&self) -> KWResult<Self::AdminClient>

fn get_topics<'life0, 'async_trait, C>( &'life0 self, ) -> Pin<Box<dyn Future<Output = KWResult<Vec<MetadataTopicWrap>>> + Send + 'async_trait>>

where C: 'async_trait + ClientContext, Self: Sync + 'async_trait, 'life0: 'async_trait,