Allow getting option values providing a template type

include/cppkafka/configuration.h

@@ -75,6 +75,7 @@ public:
     using SocketCallback = std::function<int(int domain, int type, int protoco)>;
 
     using ConfigurationBase<Configuration>::set;
+    using ConfigurationBase<Configuration>::get;
 
     /**
      * Default constructs a Configuration object

include/cppkafka/configuration_base.h

@@ -32,11 +32,15 @@
 
 #include <string>
 #include <map>
+#include "exceptions.h"
 
 namespace cppkafka {
 
 template <typename Concrete>
 class ConfigurationBase {
+private:
+    template <typename T>
+    struct Type2Type { };
 public:
     /**
      * Sets a bool value
@@ -60,6 +64,25 @@ public:
     void set(const std::string& name, const char* value) {
         proxy_set(name, value);
     }
+
+    /**
+     * \brief Gets a value, converting it to the given type.
+     *
+     * If the configuration option is not found, then ConfigOptionNotFound is thrown.
+     *
+     * If the configuration value can't be converted to the given type, then 
+     * InvalidConfigOptionType is thrown.
+     *
+     * Valid conversion types:
+     * * std::string
+     * * bool
+     * * int
+     */
+    template <typename T>
+    T get(const std::string& name) const {
+        std::string value = static_cast<const Concrete&>(*this).get(name);
+        return convert(value, Type2Type<T>());
+    }
 protected:
     static std::map<std::string, std::string> parse_dump(const char** values, size_t count) {
         std::map<std::string, std::string> output;
@@ -72,6 +95,31 @@ private:
     void proxy_set(const std::string& name, const std::string& value) {
         static_cast<Concrete&>(*this).set(name, value);
     }
+
+    static std::string convert(const std::string& value, Type2Type<std::string>) {
+        return value;
+    }
+
+    static bool convert(const std::string& value, Type2Type<bool>) {
+        if (value == "true") {
+            return true;
+        }
+        else if (value == "false") {
+            return false;
+        }
+        else {
+            throw InvalidConfigOptionType(value, "bool");
+        }
+    }
+
+    static int convert(const std::string& value, Type2Type<int>) {
+        try {
+            return std::stoi(value);
+        }
+        catch (std::exception&) {
+            throw InvalidConfigOptionType(value, "int");
+        }
+    }
 };
 
 } // cppkafka

include/cppkafka/exceptions.h

@@ -65,6 +65,14 @@ public:
     ConfigOptionNotFound(const std::string& config_name);
 };
 
+/** 
+ * Indicates a configuration option value could not be converted to a specified type
+ */
+class CPPKAFKA_API InvalidConfigOptionType : public Exception {
+public:
+    InvalidConfigOptionType(const std::string& config_name, const std::string& type);
+};
+
 /**
  * A generic rdkafka handle error
  */

include/cppkafka/topic_configuration.h

@@ -64,6 +64,7 @@ public:
                                                       int32_t partition_count)>;
 
     using ConfigurationBase<TopicConfiguration>::set;
+    using ConfigurationBase<TopicConfiguration>::get;
 
     /**
      * Default constructs a topic configuration object

src/exceptions.cpp

@@ -58,6 +58,13 @@ ConfigOptionNotFound::ConfigOptionNotFound(const string& config_name)
 
 }
 
+// InvalidConfigOptionType
+
+InvalidConfigOptionType::InvalidConfigOptionType(const string& config_name, const string& type) 
+: Exception(config_name + " could not be converted to " + type) {
+
+}
+
 // HandleException
 
 HandleException::HandleException(rd_kafka_resp_err_t error_code) 

tests/configuration_test.cpp

@@ -4,6 +4,8 @@
 
 using namespace cppkafka;
 
+using std::string;
+
 class ConfigurationTest : public testing::Test {
 public:
     
@@ -13,6 +15,7 @@ TEST_F(ConfigurationTest, GetSetConfig) {
     Configuration config;
     config.set("group.id", "foo");
     EXPECT_EQ("foo", config.get("group.id"));
+    EXPECT_EQ("foo", config.get<string>("group.id"));
 
     EXPECT_THROW(config.get("asd"), ConfigOptionNotFound);
 }
@@ -21,6 +24,7 @@ TEST_F(ConfigurationTest, GetSetTopicConfig) {
     TopicConfiguration config;
     config.set("auto.commit.enable", true);
     EXPECT_EQ("true", config.get("auto.commit.enable"));
+    EXPECT_EQ(true, config.get<bool>("auto.commit.enable"));
 
     EXPECT_THROW(config.get("asd"), ConfigOptionNotFound);
 }
@@ -32,6 +36,7 @@ TEST_F(ConfigurationTest, SetOverloads) {
 
     EXPECT_EQ("true", config.get("enable.auto.commit"));
     EXPECT_EQ("100", config.get("auto.commit.interval.ms"));
+    EXPECT_EQ(100, config.get<int>("auto.commit.interval.ms"));
 }
 
 TEST_F(ConfigurationTest, GetAll) {