scottk/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2025-11-04 20:48:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make BasicConsumerDispatcher::Pauser a template type

Browse Source
This commit is contained in:
Matias Fontanini
2017-08-21 14:47:06 -07:00
parent 0d4b9ef2f6
commit 86d4bc8037
1 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
include/cppkafka/utils/consumer_dispatcher.h
View File

@@ -145,9 +145,10 @@ private:
} }
// Simple RAII wrapper for pausing/resuming // Simple RAII wrapper for pausing/resuming
template <typename C>
class Pauser { class Pauser {
public: public:
Pauser(Consumer& consumer, const TopicPartitionList& topic_partitions) Pauser(C& consumer, const TopicPartitionList& topic_partitions)
: consumer_(consumer), topic_partitions_(topic_partitions) { : consumer_(consumer), topic_partitions_(topic_partitions) {
consumer_.pause_partitions(topic_partitions_); consumer_.pause_partitions(topic_partitions_);
} }
@@ -159,7 +160,7 @@ private:
Pauser(const Pauser&) = delete; Pauser(const Pauser&) = delete;
Pauser& operator=(const Pauser&) = delete; Pauser& operator=(const Pauser&) = delete;
private: private:
Consumer& consumer_; C& consumer_;
TopicPartitionList topic_partitions_; TopicPartitionList topic_partitions_;
}; };
@@ -299,7 +300,7 @@ private:
// The callback rejected the message, start throttling // The callback rejected the message, start throttling
if (msg) { if (msg) {
// Pause consumption. When the pauser goes off scope, it will resume it // Pause consumption. When the pauser goes off scope, it will resume it
Pauser pauser(consumer_, consumer_.get_assignment()); Pauser<ConsumerType> pauser(consumer_, consumer_.get_assignment());
// Handle throttling on this message // Handle throttling on this message
on_throttle(Throttle{}, callback, std::move(msg)); on_throttle(Throttle{}, callback, std::move(msg));