Struct kafka_threadpool::kafka_publisher::KafkaPublisher

source · [−]

pub struct KafkaPublisher {
    pub config: KafkaClientConfig,
    pub publish_msgs: Arc<Mutex<Vec<KafkaPublishMessage>>>,
}

Expand description

KafkaPublishMessage

API object for clients calling [start_threadpool]

config - holds the static configuration for each thread (connectivity endpoints, tls assets, etc.)
publish_msgs - lockable work Vec that is shared by any thread(s) that want to publish KafkaPublishMessage messages to Kafka

Fields

config: KafkaClientConfigpublish_msgs: Arc<Mutex<Vec<KafkaPublishMessage>>>

Implementations

source

impl KafkaPublisher

source

pub fn new() -> Self

new

create a new singleton KafkaPublisher for interfacing with the backend kafka publish threadpool

pub fn is_enabled(&self) -> bool

is_enabled

Clients wanting to test if the threadpool is enabled can use this helper method.

Returns

true when the environment variable: KAFKA_ENABLED is set to true or 1
false when the environment variable: KAFKA_ENABLED is set to anything but true or 1

source

pub async fn add_data_msg(
 &self,
 topic: &str,
 key: &str,
 headers: Option<HashMap<String, String>>,
 payload: &str
) -> Result<usize, String>

add_data_msg

Build a publishable data message from function arguments and add it to the lockable publish vector. Client libraries that want to just send a single message without worrying about the KafkaPublishMessageType should use this function.

Arguments

topic - kafka topic to publish the message into
key - kafka partition key
headers - optional - headers for the kafka message
payload - data within the kafka message

Uses the utility API method: add_messages_to_locked_work_vec

Returns

Result<usize, String> where

usize = updated number of messages in self.publish_msgs after adding the new msg
String = error reason

source

pub async fn add_msg(&self, msg: KafkaPublishMessage) -> Result<usize, String>

add_msg

Add a single message to the lockable publish vector

Arguments

msg - an initialized KafkaPublishMessage to add to the lockable work vector: self.publish_msgs

Uses the utility API method: add_messages_to_locked_work_vec

Returns

Result<usize, String> where

usize = updated number of messages in self.publish_msgs after adding the new msg
String = error reason

source

pub async fn add_msgs(
&self,
msgs: Vec<KafkaPublishMessage>
) -> Result<usize, String>

add_msgs

Add a vector of messages to the lockable publish vector

Arguments

msgs - vector of KafkaPublishMessage to add to the lockable work vector: self.publish_msgs

Uses the utility API method: add_messages_to_locked_work_vec

Returns

Result<usize, String> where

usize = updated number of messages in self.publish_msgs after adding the new msgs
String = error reason

source

pub async fn drain_msgs(&self) -> Vec<KafkaPublishMessage>

drain_msgs

Helper function for testing - allows draining all data in the lockable work vec: self.publish_msgs

Returns

Vec<KafkaPublishMessage> containing all drained messages

source

pub async fn shutdown(&self) -> Result<String, String>

shutdown

Gracefully shutdown the threadpool by sending the Shutdown control message to all worker threads

Errors

Threads may get hung if something goes wrong.

Examples

my_threadpool.shutdown().await.unwrap();

source

pub async fn get_metadata(&self, fetch_offsets: bool, topic: Option<&str>)

get_metadata

Get kafka cluster information by all topics or for just one topic

Arguments

fetch_offsets - when true this function will count the total number of messages in each topic
topic - If set, only get the details for that specific topic if set to None get details for all topics