First real test using MQTT to send and receive messages.

- also try to setup CI (might fail)

.gitlab-ci.yml

 variables:
   GIT_SUBMODULE_STRATEGY: recursive
 
+services:
+  - name: "eclipse-mosquitto:1.6.9"
+    alias: "mqtt"
+
 stages:
   - build
 
@@ -8,14 +12,6 @@ build:
   image: openjdk:8
   stage: build
   before_script:
-    # - git submodule init
-    # - git submodule sync
-    # - cat .gitmodules
-    # - cat .git/config
-    # - ls -lah *
-    # - rm -rf relast.preprocessor/*
-    # - ls -lah *
-    # - git submodule update --remote
     - ls -lah *
   script:
     - ./gradlew --no-daemon build

ros2rag.base/src/main/resources/MqttUpdater.jadd

@@ -145,6 +145,7 @@ public class MqttUpdater {
 
     // subscribe at broker
     org.fusesource.mqtt.client.Topic[] topicArray = { new org.fusesource.mqtt.client.Topic(topic, this.qos) };
+    connection.getDispatchQueue().execute(() -> {
       connection.subscribe(topicArray, new org.fusesource.mqtt.client.Callback<byte[]>() {
         @Override
         public void onSuccess(byte[] qoses) {
@@ -156,6 +157,7 @@ public class MqttUpdater {
           logger.error("Could not subscribe to {}", topic, cause);
         }
       });
+    });
   }
 
   /**
@@ -187,6 +189,7 @@ public class MqttUpdater {
       logger.warn("Stopping without connection. Was setHost() called?");
       return;
     }
+    connection.getDispatchQueue().execute(() -> {
       connection.disconnect(new org.fusesource.mqtt.client.Callback<Void>() {
         @Override
         public void onSuccess(Void value) {
@@ -198,9 +201,11 @@ public class MqttUpdater {
           // Disconnects never fail. And we do not care either.
         }
       });
+    });
   }
 
   public void publish(String topic, byte[] bytes) {
+    connection.getDispatchQueue().execute(() -> {
       connection.publish(topic, bytes, qos, false, new org.fusesource.mqtt.client.Callback<Void>() {
         @Override
         public void onSuccess(Void value) {
@@ -212,5 +217,6 @@ public class MqttUpdater {
           logger.warn("Could not publish on topic '{}'", topic);
         }
       });
+    });
   }
 }

ros2rag.tests/build.gradle

@@ -89,6 +89,7 @@ task preprocessExampleTest(type: JavaExec, group: 'verification') {
 task compileExampleTest(type: RelastTest) {
     verbose = true
     useJastAddNames = true
+    jastAddList = 'JastAddList'
     relastFiles 'src/test/02-after-ros2rag/example/Grammar.relast'
     grammarName = 'src/test/03-after-relast/example/example'
     packageName = 'example.ast'

ros2rag.tests/src/test/01-input/example/Example.jadd

@@ -79,7 +79,10 @@ aspect GrammarTypes {
     return false;
   }
 
+  syn double RobotArm.speedLow() = 0.4d;
+  syn double RobotArm.speedHigh() = 1.0d;
+
   syn double RobotArm.getAppropriateSpeed() {
-    return isInSafetyZone() ? 0.4d : 1.0d;
+    return isInSafetyZone() ? speedLow() : speedHigh();
   }
 }

ros2rag.tests/src/test/java/org/jastadd/ros2rag/tests/ExampleTest.java

 package org.jastadd.ros2rag.tests;
 
+import com.google.protobuf.InvalidProtocolBufferException;
+import config.Robotconfig.RobotConfig;
 import example.ast.*;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.*;
 
 /**
  * Test case "example".
@@ -12,10 +21,90 @@ import java.io.IOException;
  */
 public class ExampleTest {
 
+  private static final double DELTA = 0.001d;
+  private static final String TOPIC_CONFIG = "robot/config";
+  private static final String TOPIC_JOINT1 = "robot/arm/joint1";
+
+  private Model model;
+  private RobotArm robotArm;
+  private Joint joint1;
+
+  private CountDownLatch stop;
+
+  @AfterEach
+  public void signalCondition() {
+    if (stop != null) {
+      stop.countDown();
+    }
+  }
+
+  @Test
+  public void buildModel() {
+    createModel();
+  }
+
   @Test
-  public void buildModel() throws IOException {
-    Model model = new Model();
-    model.MqttSetHost("localhost");
+  public void communicate() throws IOException, InterruptedException {
+    createModel();
+    List<RobotConfig> receivedConfigs = new ArrayList<>();
+
+    CountDownLatch start = new CountDownLatch(1);
+    stop = new CountDownLatch(1);
+    createListenerThread(receivedConfigs, start, stop);
+    start.await();
+
+    model.MqttSetHost(TestUtils.getMqttHost());
+    assertTrue(model.MqttWaitUntilReady(2, TimeUnit.SECONDS));
+    // joint is currently within the safety zone, so speed should be low
+    robotArm.connectAppropriateSpeed(TOPIC_CONFIG, true);
+    joint1.connectCurrentPosition(TOPIC_JOINT1);
+
+    // now change the position of the joint out of the safety zone
+    joint1.setCurrentPosition(makePosition(2, 2, 2));
+
+    // and wait for MQTT to send/receive messages
+    TimeUnit.SECONDS.sleep(2);
+
+    // there should be two configs received by now (the initial one, and the updated)
+    assertEquals(2, receivedConfigs.size());
+    RobotConfig actualInitialConfig = receivedConfigs.get(0);
+    assertEquals(robotArm.speedLow(), actualInitialConfig.getSpeed(), DELTA);
+
+    RobotConfig actualUpdatedConfig = receivedConfigs.get(1);
+    assertEquals(robotArm.speedHigh(), actualUpdatedConfig.getSpeed(), DELTA);
+    model.MqttCloseConnections();
+  }
+
+  private void createListenerThread(List<RobotConfig> receivedConfigs,
+                                    CountDownLatch startCondition, CountDownLatch stopCondition) {
+    new Thread(() -> {
+      MqttUpdater receiver = new MqttUpdater("receiver");
+      try {
+        receiver.setHost(TestUtils.getMqttHost(), TestUtils.getMqttDefaultPort());
+      } catch (IOException e) {
+        fail("Could not set host: " + e.getMessage());
+      }
+      assertTrue(receiver.waitUntilReady(2, TimeUnit.SECONDS));
+      receiver.newConnection(TOPIC_CONFIG, bytes -> {
+        try {
+          RobotConfig config = RobotConfig.parseFrom(bytes);
+          receivedConfigs.add(config);
+        } catch (InvalidProtocolBufferException e) {
+          fail("Received bad config: " + e.getMessage());
+        }
+      });
+      startCondition.countDown();
+      try {
+        stopCondition.await();
+        receiver.close();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }).start();
+  }
+
+  private void createModel() {
+    model = new Model();
 
     ZoneModel zoneModel = new ZoneModel();
     zoneModel.setSize(makePosition(1, 1, 1));
@@ -32,10 +121,10 @@ public class ExampleTest {
     zoneModel.addSafetyZone(safetyZone);
     model.setZoneModel(zoneModel);
 
-    RobotArm robotArm = new RobotArm();
+    robotArm = new RobotArm();
     robotArm.setAttributeTestSource(1);  // set initial value, no trigger
 
-    Joint joint1 = new Joint();
+    joint1 = new Joint();
     joint1.setName("joint1");
     joint1.setCurrentPosition(myPosition);
 

ros2rag.tests/src/test/java/org/jastadd/ros2rag/tests/TestUtils.java

+package org.jastadd.ros2rag.tests;
+
+/**
+ * Utility methods for tests.
+ *
+ * @author rschoene - Initial contribution
+ */
+public class TestUtils {
+
+  public static String getMqttHost() {
+    if (System.getenv("GITLAB_CI") != null) {
+      // we are in the CI, so use "mqtt" as host
+      return "mqtt";
+    } {
+      // else assume a locally running mqtt broker
+      return "localhost";
+    }
+  }
+
+  public static int getMqttDefaultPort() {
+    return 1883;
+  }
+
+}