Add error handler to compacted topic processor

This commit is contained in:
Matias Fontanini
2016-07-03 08:32:13 -07:00
parent cf6ac1675b
commit 065f899954

View File

@@ -133,6 +133,11 @@ public:
*/
using EventHandler = std::function<void(Event)>;
/**
* Callback used for error handling
*/
using ErrorHandler = std::function<void(Message)>;
/**
* \brief Constructs an instance
*/
@@ -159,6 +164,11 @@ public:
*/
void set_event_handler(EventHandler callback);
/**
* \brief Sets the error handler callback
*/
void set_error_handler(ErrorHandler callback);
/**
* \brief Processes the next event
*/
@@ -170,6 +180,7 @@ private:
KeyDecoder key_decoder_;
ValueDecoder value_decoder_;
EventHandler event_handler_;
ErrorHandler error_handler_;
std::map<TopicPartition, int64_t> partition_offsets_;
Consumer::AssignmentCallback original_assignment_callback_;
};
@@ -255,11 +266,16 @@ void CompactedTopicProcessor<K, V>::set_event_handler(EventHandler callback) {
event_handler_ = std::move(callback);
}
template <typename K, typename V>
void CompactedTopicProcessor<K, V>::set_error_handler(ErrorHandler callback) {
error_handler_ = std::move(callback);
}
template <typename Key, typename Value>
void CompactedTopicProcessor<Key, Value>::process_event() {
Message message = consumer_.poll();
if (message) {
if (!message.has_error()) {
if (!message.get_error()) {
Key key = key_decoder_(message.get_key());
if (message.get_payload()) {
// If there's a payload, generate a SET_ELEMENT event
@@ -281,6 +297,9 @@ void CompactedTopicProcessor<Key, Value>::process_event() {
event_handler_({ Event::REACHED_EOF, message.get_topic(),
message.get_partition() });
}
else if (error_handler_) {
error_handler_(std::move(message));
}
}
}
}