1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374 |
- #ifndef __KAFKA_CLIENT_H__
- #define __KAFKA_CLIENT_H__
- #include <librdkafka/rdkafka.h>
- #include <stdarg.h>
- #include <functional>
- #include <string>
- namespace cnstream {
- class KafkaClient {
- public:
- enum class TYPE {
- PRODUCER = 0,
- CONSUMER,
- };
- explicit KafkaClient(TYPE type, const std::string &brokers, const std::string &topic, int32_t partition);
- ~KafkaClient();
- bool Start();
- bool Stop(bool instant = false);
- bool Produce(const uint8_t *p_payload, size_t length);
- bool Consume(uint8_t **p_payload, size_t *p_length, int timeout_ms = 0);
- private:
- enum class STATE {
- IDLE = 0,
- PRE_PRODUCE,
- PRE_CONSUME,
- PRODUCE,
- CONSUME,
- };
- static void logger(const rd_kafka_t *rk, int level, const char *fac, const char *buf);
- static void msg_delivered(rd_kafka_t *rk, const rd_kafka_message_t *msg, void *opaque);
- bool msg_consume(rd_kafka_message_t *msg, uint8_t **p_payload, size_t *p_len);
- TYPE type_;
- const std::string brokers_;
- const std::string topic_;
- int32_t partition_ = 0;
- STATE state_ = STATE::IDLE;
- rd_kafka_t *rk_ = nullptr;
- rd_kafka_topic_t *rkt_ = nullptr;
- rd_kafka_conf_t *conf_ = nullptr;
- rd_kafka_topic_conf_t *topic_conf_ = nullptr;
- rd_kafka_message_t *message_ = nullptr;
- };
- }
- #endif
|