feat(node-kafka): Port kafka node-type to C++ and simplify it

Signed-off-by: Steffen Vogel <[email protected]>

Author: stv0g

CMakeLists.txt

@@ -103,7 +103,7 @@ pkg_check_modules(LIB60870 IMPORTED_TARGET lib60870>=2.3.1)
 pkg_check_modules(LIBCONFIG IMPORTED_TARGET libconfig>=1.4.9)
 pkg_check_modules(MOSQUITTO IMPORTED_TARGET libmosquitto>=1.6.9)
 pkg_check_modules(MODBUS IMPORTED_TARGET libmodbus>=3.1.0)
-pkg_check_modules(RDKAFKA IMPORTED_TARGET rdkafka>=1.5.0)
+pkg_check_modules(RDKAFKAPP IMPORTED_TARGET rdkafka++>=1.5.0)
 pkg_check_modules(HIREDIS IMPORTED_TARGET hiredis>=1.0.0)
 pkg_check_modules(REDISPP IMPORTED_TARGET redis++>=1.2.0)
 pkg_check_modules(RABBITMQ_C IMPORTED_TARGET librabbitmq>=0.8.0)
@@ -193,7 +193,7 @@ cmake_dependent_option(WITH_NODE_IEC60870       "Build with iec60870 node-types"
 cmake_dependent_option(WITH_NODE_IEC61850       "Build with iec61850 node-types"                        "${WITH_DEFAULTS}" "LIBIEC61850_FOUND; NOT WITHOUT_GPL" OFF)
 cmake_dependent_option(WITH_NODE_INFINIBAND     "Build with infiniband node-type"                       "${WITH_DEFAULTS}" "IBVerbs_FOUND; RDMACM_FOUND" OFF) # Infiniband node-type is currenly broken
 cmake_dependent_option(WITH_NODE_INFLUXDB       "Build with influxdb node-type"                         "${WITH_DEFAULTS}" "" OFF)
-cmake_dependent_option(WITH_NODE_KAFKA          "Build with kafka node-type"                            "${WITH_DEFAULTS}" "RDKAFKA_FOUND" OFF)
+cmake_dependent_option(WITH_NODE_KAFKA          "Build with kafka node-type"                            "${WITH_DEFAULTS}" "RDKAFKAPP_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_LOOPBACK       "Build with loopback node-type"                         "${WITH_DEFAULTS}" "" OFF)
 cmake_dependent_option(WITH_NODE_MODBUS         "Build with modbus node-type"                           "${WITH_DEFAULTS}" "MODBUS_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_MQTT           "Build with mqtt node-type"                             "${WITH_DEFAULTS}" "MOSQUITTO_FOUND" OFF)
@@ -309,7 +309,7 @@ add_feature_info(NODE_MODBUS            WITH_NODE_MODBUS            "Build with
 add_feature_info(NODE_MQTT              WITH_NODE_MQTT              "Build with mqtt node-type")
 add_feature_info(NODE_NANOMSG           WITH_NODE_NANOMSG           "Build with nanomsg node-type")
 add_feature_info(NODE_NGSI              WITH_NODE_NGSI              "Build with ngsi node-type")
-add_feature_info(NODE_OPAL_AYSNC        WITH_NODE_OPAL_ASYNC        "Build with opal.async node-type")
+add_feature_info(NODE_OPAL_ASYNC        WITH_NODE_OPAL_ASYNC        "Build with opal.async node-type")
 add_feature_info(NODE_OPAL_ORCHESTRA    WITH_NODE_OPAL_ORCHESTRA    "Build with opal.orchestra node-type")
 add_feature_info(NODE_OPENDSS           WITH_NODE_OPENDSS           "Build with opendss node-type")
 add_feature_info(NODE_REDIS             WITH_NODE_REDIS             "Build with redis node-type")

doc/openapi/components/schemas/config/nodes/kafka.yaml

@@ -56,7 +56,7 @@ allOf:
     in:
       type: object
       properties:
-        consume:
+        topic:
           type: string
           description: The Kafka topic to which this node-type will subscribe for receiving messages.
 
@@ -67,7 +67,7 @@ allOf:
     out:
       type: object
       properties:
-        produce:
+        topic:
           type: string
           description: The Kafka topic to which this node-type will publish messages.
 

etc/examples/nodes/kafka.conf

@@ -7,27 +7,39 @@ nodes = {
 
         format = "json.kafka"
 
-        server = "localhost:9094"
-        protocol = "SASL_SSL"
+        server = "localhost:9092"
+        protocol = "PLAINTEXT"
         client_id = "villas-node"
 
         in = {
-            consume = "test-topic"
+            topic = "test-topic"
             group_id = "villas-node"
         }
 
         out = {
-            produce = "test-topic"
+            topic = "test-topic"
         }
+    }
 
-        ssl = {
-            ca = "/etc/ssl/certs/ca.pem"
-        }
+    siggen = {
+        type = "signal"
 
-        sasl = {
-            mechanisms = "SCRAM-SHA-512"
-            username = "scram-sha-512-usr"
-            password = "scram-sha-512-pwd"
-        }
+        rate = 20
+        values = 5
+        signal = "mixed"
     }
 }
+
+paths = (
+    {
+        in = "siggen"
+        out = "kafka_node"
+    },
+    {
+        in = "kafka_node"
+
+        hooks = [
+            "print"
+        ]
+    }
+)

lib/nodes/CMakeLists.txt

@@ -123,7 +123,7 @@ endif()
 # Enable Kafka support
 if(WITH_NODE_KAFKA)
     list(APPEND NODE_SRC kafka.cpp)
-    list(APPEND LIBRARIES  PkgConfig::RDKAFKA)
+    list(APPEND LIBRARIES PkgConfig::RDKAFKAPP)
 endif()
 
 # Enable Comedi support