fixing partition selector, validator, and serializer

include/mofka/PartitionSelector.hpp

@@ -146,7 +146,9 @@ class PartitionSelector {
      *
      * @return PartitionSelector instance.
      */
-    static PartitionSelector FromMetadata(const char* type, const Metadata& metadata);
+    static PartitionSelector FromMetadata(
+            const char* type,
+            const Metadata& metadata);
 
     /**
      * @brief Version of the above function that does not require a type.

include/mofka/Serializer.hpp

@@ -147,7 +147,9 @@ class Serializer {
      *
      * @return Serializer instance.
      */
-    static Serializer FromMetadata(const char* type, const Metadata& metadata);
+    static Serializer FromMetadata(
+            const char* type,
+            const Metadata& metadata);
 
     /**
      * @brief Same as the above function but the type is expected

include/mofka/Validator.hpp

@@ -136,7 +136,9 @@ class Validator {
      *
      * @return Validator instance.
      */
-    static Validator FromMetadata(const char* type, const Metadata& metadata);
+    static Validator FromMetadata(
+            const char* type,
+            const Metadata& metadata);
 
     /**
      * @brief Same as the above function but will look for a "__type__"

python/mochi/mofka/client.py

@@ -44,6 +44,11 @@ def __init__(self, group_file, arg=None):
         else:
             super().__init__(group_file, self._mid)
 
+    def create_topic(self, *args, schema: dict|None = None, **kwargs):
+        if schema is not None:
+            kwargs["validator"] = Validator.from_metadata("schema", {"schema":schema})
+        super().create_topic(*args, **kwargs)
+
 
 class ServiceHandle(MofkaDriver):
 

python/py-mofka-client.cpp

@@ -110,14 +110,14 @@ PYBIND11_MODULE(pymofka_client, m) {
     ;
 
     py::class_<mofka::PartitionSelector>(m, "PartitionSelector")
-        .def_static("from_metadata",
-            [](const nlohmann::json& md){
-                return mofka::PartitionSelector::FromMetadata(md);
-            }, "metadata"_a=nlohmann::json::object())
         .def_static("from_metadata",
             [](const char* type, const nlohmann::json& md){
                 return mofka::PartitionSelector::FromMetadata(type, md);
             }, "type"_a, "metadata"_a=nlohmann::json::object())
+        .def_static("from_metadata",
+            [](const nlohmann::json& md){
+                return mofka::PartitionSelector::FromMetadata(md);
+            }, "metadata"_a=nlohmann::json::object())
     ;
 
     py::class_<mofka::MofkaDriver>(m, "MofkaDriver")

src/BatchProducer.hpp

@@ -54,9 +54,7 @@ class BatchProducer : public ProducerInterface {
     , m_topic(std::move(topic))
     {}
 
-    ~BatchProducer() {
-        flush();
-    }
+    virtual ~BatchProducer() = default;
 
     const std::string& name() const override {
         return m_name;

src/EventbridgeValidator.cpp

@@ -534,7 +534,7 @@ void EventbridgeValidator::validate(const Metadata& metadata, const Data& data)
 
 Metadata EventbridgeValidator::metadata() const {
     json config = json::object();
-    config["__type__"] = "eventbridge";
+    config["type"] = "eventbridge";
     config["schema"] = m_schema;
     return Metadata{std::move(config)};
 }

src/MofkaProducer.hpp

@@ -43,6 +43,10 @@ class MofkaProducer : public BatchProducer {
                   std::shared_ptr<MofkaTopicHandle> topic);
 
     std::shared_ptr<ProducerBatchInterface> newBatchForPartition(size_t index) const override;
+
+    ~MofkaProducer() {
+        flush();
+    }
 };
 
 }

src/PartitionSelector.cpp

@@ -45,14 +45,14 @@ PartitionSelector PartitionSelector::FromMetadata(const Metadata& metadata) {
             "Cannot create PartitionSelector from Metadata: "
             "invalid Metadata (expected JSON object)");
     }
-    if(!json.contains("__type__")) {
+    if(!json.contains("type")) {
         return PartitionSelector{};
     }
-    auto& type = json["__type__"];
+    auto& type = json["type"];
     if(!type.is_string()) {
         throw Exception(
             "Cannot create PartitionSelector from Metadata: "
-            "invalid __type__ in Metadata (expected string)");
+            "invalid \"type\" field in Metadata (expected string)");
     }
     auto& type_str = type.get_ref<const std::string&>();
     std::shared_ptr<PartitionSelectorInterface> ts = PartitionSelectorFactory::create(type_str, metadata);
@@ -67,7 +67,7 @@ PartitionSelector PartitionSelector::FromMetadata(const char* type, const Metada
             "invalid Metadata (expected JSON object)");
     }
     auto md_copy = metadata;
-    md_copy.json()["__type__"] = type;
+    md_copy.json()["type"] = type;
     std::shared_ptr<PartitionSelectorInterface> ts = PartitionSelectorFactory::create(type, md_copy);
     return ts;
 }

src/Serializer.cpp

@@ -44,14 +44,14 @@ Serializer Serializer::FromMetadata(const Metadata& metadata) {
                 "Cannot create Serializer from Metadata: "
                 "invalid Metadata (expected JSON object)");
     }
-    if(!json.contains("__type__")) {
+    if(!json.contains("type")) {
         return Serializer{};
     }
-    auto& type = json["__type__"];
+    auto& type = json["type"];
     if(!type.is_string()) {
         throw Exception(
                 "Cannot create Serializer from Metadata: "
-                "invalid __type__ in Metadata (expected string)");
+                "invalid \"type\" field in Metadata (expected string)");
     }
     const auto& type_str = type.get_ref<const std::string&>();
     std::shared_ptr<SerializerInterface> s = SerializerFactory::create(type_str, metadata);
@@ -66,7 +66,7 @@ Serializer Serializer::FromMetadata(const char* type, const Metadata& metadata)
                 "invalid Metadata (expected JSON object)");
     }
     auto md_copy = metadata;
-    md_copy.json()["__type__"] = type;
+    md_copy.json()["type"] = type;
     std::shared_ptr<SerializerInterface> s = SerializerFactory::create(type, md_copy);
     return Serializer(std::move(s));
 }

src/Validator.cpp

@@ -40,14 +40,14 @@ Validator Validator::FromMetadata(const Metadata& metadata) {
             "Cannot create Validator from Metadata: "
             "invalid Metadata (expected JSON object)");
     }
-    if(!json.contains("__type__")) {
+    if(!json.contains("type")) {
         return Validator{};
     }
-    auto& type = json["__type__"];
+    auto& type = json["type"];
     if(!type.is_string()) {
         throw Exception(
             "Cannot create Validator from Metadata: "
-            "invalid __type__ in Metadata (expected string)");
+            "invalid \"type\" field in Metadata (expected string)");
     }
     auto& type_str = type.get_ref<const std::string&>();
     std::shared_ptr<ValidatorInterface> v = ValidatorFactory::create(type_str, metadata);
@@ -62,7 +62,7 @@ Validator Validator::FromMetadata(const char* type, const Metadata& metadata) {
             "invalid Metadata (expected JSON object)");
     }
     auto md_copy = metadata;
-    md_copy.json()["__type__"] = type;
+    md_copy.json()["type"] = type;
     std::shared_ptr<ValidatorInterface> v = ValidatorFactory::create(type, md_copy);
     return v;
 }

tests/python/customs/MyPartitionSelector.cpp

@@ -14,9 +14,10 @@ class MyPartitionSelector : public mofka::PartitionSelectorInterface {
         (void)metadata;
         if(m_targets.size() == 0)
             throw mofka::Exception("PartitionSelector has no target to select from");
-        if(m_index >= m_targets.size()) m_index = m_index % m_targets.size();
+        auto result = m_index;
         m_index += 1;
-        return m_targets.at(m_index-1);
+        if(m_index == m_targets.size()) m_index = 0;
+        return result;
     }
 
     mofka::Metadata metadata() const override {

tests/python/test_mofka_driver.py

@@ -31,24 +31,18 @@ def test_get_servers(self):
     def test_create_open_topic(self):
         """Test create and open a topic"""
         name = "my_topic"
-        validator = mofka.Validator.from_metadata(
-            {"__type__":"my_validator:libmy_validator.so"})
-        selector = mofka.PartitionSelector.from_metadata(
-            {"__type__":"my_partition_selector:libmy_partition_selector.so"})
-        serializer = mofka.Serializer.from_metadata(
-            {"__type__":"my_serializer:libmy_serializer.so"})
+        validator = mofka.Validator.from_metadata("my_validator:libmy_validator.so")
+        selector = mofka.PartitionSelector.from_metadata("my_partition_selector:libmy_partition_selector.so")
+        serializer = mofka.Serializer.from_metadata("my_serializer:libmy_serializer.so")
         self.service.create_topic(name, validator, selector, serializer)
         topic = self.service.open_topic(name)
 
     def test_add_memory_partition(self):
         """Test add partition"""
         topic_name = "my_topic"
-        validator = mofka.Validator.from_metadata(
-            {"__type__":"my_validator:libmy_validator.so"})
-        selector = mofka.PartitionSelector.from_metadata(
-            {"__type__":"my_partition_selector:libmy_partition_selector.so"})
-        serializer = mofka.Serializer.from_metadata(
-            {"__type__":"my_serializer:libmy_serializer.so"})
+        validator = mofka.Validator.from_metadata("my_validator:libmy_validator.so")
+        selector = mofka.PartitionSelector.from_metadata("my_partition_selector:libmy_partition_selector.so")
+        serializer = mofka.Serializer.from_metadata("my_serializer:libmy_serializer.so")
         server_rank = 0
         self.service.create_topic(topic_name, validator, selector, serializer)
         self.service.add_memory_partition(topic_name, server_rank)

tests/python/test_mofka_producer.py

@@ -33,12 +33,9 @@ def setUp(self):
 
         # create a topic
         name = "my_topic"
-        validator = mofka.Validator.from_metadata(
-            {"__type__":"my_validator:libmy_validator.so"})
-        selector = mofka.PartitionSelector.from_metadata(
-            {"__type__":"my_partition_selector:libmy_partition_selector.so"})
-        serializer = mofka.Serializer.from_metadata(
-            {"__type__":"my_serializer:libmy_serializer.so"})
+        validator = mofka.Validator.from_metadata("my_validator:libmy_validator.so")
+        selector = mofka.PartitionSelector.from_metadata("my_partition_selector:libmy_partition_selector.so")
+        serializer = mofka.Serializer.from_metadata("my_serializer:libmy_serializer.so")
         self.service.create_topic(name, validator, selector, serializer)
         self.service.add_memory_partition(name, 0)
 

tests/python/test_mofka_topic_handle.py

@@ -20,12 +20,9 @@ def setUp(self):
         self.service = mofka.MofkaDriver("mofka.json", self.bedrock_server.margo.mid)
 
         name = "my_topic"
-        validator = mofka.Validator.from_metadata(
-            {"__type__":"my_validator:libmy_validator.so"})
-        selector = mofka.PartitionSelector.from_metadata(
-            {"__type__":"my_partition_selector:libmy_partition_selector.so"})
-        serializer = mofka.Serializer.from_metadata(
-            {"__type__":"my_serializer:libmy_serializer.so"})
+        validator = mofka.Validator.from_metadata("my_validator:libmy_validator.so")
+        selector = mofka.PartitionSelector.from_metadata("my_partition_selector:libmy_partition_selector.so")
+        serializer = mofka.Serializer.from_metadata("my_serializer:libmy_serializer.so")
         self.service.create_topic(name, validator, selector, serializer)
         self.topic = self.service.open_topic(name)