added a test + Kafak setup in the Docker image

.gitlab-ci.yml

@@ -28,7 +28,7 @@ all-tests:
   tags:
     - docker-exec
   services:
-    - name: postgres:11
+    - name: postgres:13
     - name: atmoz/sftp:alpine
       command: ["foo:pass:::sftp_root"]
       alias: local-sftp
@@ -39,6 +39,7 @@ all-tests:
     when: on_failure
     expire_in: 2 days
   script:
+    - test/docker/setup.sh
     #- cp -rf SftpJobModule $OMQ_DIR/user/modules
     #- cp -rf SftpWorkflowModule $OMQ_DIR/user/modules
     #- gosu qorus:qorus oload -Rlv tests/SftpJobModule/*.q*

test/docker/setup.sh

+#!/bin/sh
+
+set -e
+set -x
+
+. /tmp/env.sh
+
+echo --- downloading Kafka
+cd /opt
+wget https://mirror.dkm.cz/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz
+echo --- installing Kafka
+tar xvf kafka_2.13-2.6.0.tgz
+cd kafka_2.13-2.6.0
+echo --- starting Kafka
+./bin/zookeeper-server-start.sh config/zookeeper.properties &
+./bin/kafka-server-start.sh config/server.properties &
+sleep 2
+echo --- creating the test topic
+bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

test/kafka/BBM_KafkaConsumerTest-1.0-service/BBM_KafkaConsumerTest.java

@@ -5,7 +5,7 @@ import org.qore.jni.QoreObject;
 import java.util.Map;
 import org.qore.jni.Hash;
 import java.lang.reflect.Method;
-import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
 
 class BBM_KafkaConsumerTest extends QorusService {
     // ==== GENERATED SECTION! DON'T EDIT! ==== //
@@ -18,9 +18,15 @@ class BBM_KafkaConsumerTest extends QorusService {
         // ======== GENERATED SECTION END ========= //
     }
 
+    static public ArrayList<Map<String, Object>> msgs = new ArrayList<Map<String, Object>>();
+
     public void init() throws Throwable {
     }
 
+    public ArrayList<Map<String, Object>> getMessages() throws Throwable {
+        return msgs;
+    }
+
     // ==== GENERATED SECTION! DON'T EDIT! ==== //
     public void start() throws Throwable {
         classConnections.start(null);
@@ -61,11 +67,7 @@ class ClassConnections_BBM_KafkaConsumerTest implements Observer { // has to inh
             return qoreObject.callMethod(methodName, params);
         } else {
             final Method method = object.getClass().getMethod(methodName, Object.class);
-            try {
-                return method.invoke(object, params);
-            } catch (InvocationTargetException ex) {
-                throw ex.getCause();
-            }
+            return method.invoke(object, params);
         }
     }
 
@@ -73,6 +75,7 @@ class ClassConnections_BBM_KafkaConsumerTest implements Observer { // has to inh
     public void update(String id, Map<String, Object> params) throws Throwable {
         if (id.equals("BBM_KafkaConsumer::eventLoop")) {
             KafkaMessageEvent(params);
+            BBM_KafkaConsumerTest.msgs.add(params);
         }
     }
 

test/kafka/BBM_KafkaConsumerTest-1.0.qsd.yaml

@@ -41,8 +41,7 @@ config-items:
         interface-version: "1.0"
   - name: kafka-topics
     value:
-      - quickstart-events
-      - new-topic
+        ["quickstart-events","new-topic"]
     parent:
         interface-type: class
         interface-name: BBM_KafkaConsumer
@@ -426,3 +425,5 @@ methods:
     desc: Start receiving messages
   - name: stop
     desc: Stop the service
+  - name: getMessages
+    desc: returns messages received

test/kafka/BBM_KafkaTest.qtest

+#!/usr/bin/env qore
+# -*- mode: qore; indent-tabs-mode: nil -*-
+
+%require-types
+%strict-args
+%new-style
+%enable-all-warnings
+
+%exec-class BBM_RestServiceBaseTest
+
+%requires QorusInterfaceTest
+
+class BBM_RestServiceBaseTest inherits QorusServiceTest {
+    constructor() : QorusServiceTest("BBM_KafkaConsumerTest", "1.0", \ARGV, Opts) {
+        addTestCase("test", \test());
+        set_return_value(main());
+    }
+
+    private usageIntern(int offset = OffsetColumn) {
+        TestReporter::usageIntern(OffsetColumn);
+    }
+
+    test() {
+        # unload consumer in case it's loaded
+        qrest.put("services/BBM_KafkaConsumerTest/unload");
+        # load consumer and producer services
+        qrest.put("services/BBM_KafkaConsumerTest/load");
+        on_exit {
+            qrest.put("services/BBM_KafkaConsumerTest/unload");
+        }
+        qrest.put("services/BBM_KafkaProducerTest/load");
+        on_exit {
+            qrest.put("services/BBM_KafkaProducerTest/unload");
+        }
+
+        hash<auto> msg = {
+            "topic": "quickstart-events",
+            "key": get_random_string(),
+            "value": get_random_string(),
+        };
+        qrest.put("services/BBM_KafkaProducerTest/sendMessage/call", {"args": msg});
+        # check messages
+        assertEq(msg, qrest.put("services/BBM_KafkaConsumerTest/getMessages/call")[0]{keys msg});
+    }
+}