Update parser and grammar according to latest discussion.

- allow multiple mappings for update definitions
- in mappings, remove target variable name and prepare for more types, e.g., arrays
- add required option to specify name of root node in Compiler
- copy MqttUpdater in compile step

build.gradle

@@ -33,7 +33,6 @@ dependencies {
 sourceSets {
     main {
         java.srcDir "src/gen/java"
-        java.srcDir "buildSrc/gen/java"
     }
 }
 

src/main/jastadd/MqttUpdater.java_class

+package org.jastadd.ros2rag.compiler.mqtt;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.fusesource.hawtbuf.Buffer;
+import org.fusesource.hawtbuf.UTF8Buffer;
+import org.fusesource.mqtt.client.*;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+
+/**
+ * Helper class to receive updates via MQTT and use callbacks to handle those messages.
+ *
+ * @author rschoene - Initial contribution
+ */
+public class MqttUpdater {
+
+  private final Logger logger;
+  private final String name;
+
+  /** The host running the MQTT broker. */
+  private URI host;
+  /** The connection to the MQTT broker. */
+  private CallbackConnection connection;
+  /** Whether we are subscribed to the topics yet */
+  private final Condition readyCondition;
+  private final Lock readyLock;
+  private boolean ready;
+  private QoS qos;
+  /** Dispatch knowledge */
+  private final Map<String, Consumer<byte[]>> callbacks;
+
+  public MqttUpdater() {
+    this("Ros2Rag");
+  }
+
+  public MqttUpdater(String name) {
+    this.name = Objects.requireNonNull(name, "Name must be set");
+    this.logger = LogManager.getLogger(MqttUpdater.class);
+    this.callbacks = new HashMap<>();
+    this.readyLock = new ReentrantLock();
+    this.readyCondition = readyLock.newCondition();
+    this.ready = false;
+    this.qos = QoS.AT_LEAST_ONCE;
+  }
+
+  /**
+   * Sets the host to receive messages from, and connects to it.
+   * @throws IOException if could not connect, or could not subscribe to a topic
+   * @return self
+   */
+  public MqttUpdater setHost(String host, int port) throws IOException {
+    this.host = URI.create("tcp://" + host + ":" + port);
+    logger.debug("Host for {} is {}", this.name, this.host);
+
+    Objects.requireNonNull(this.host, "Host need to be set!");
+    MQTT mqtt = new MQTT();
+    mqtt.setHost(this.host);
+    connection = mqtt.callbackConnection();
+    AtomicReference<Throwable> error = new AtomicReference<>();
+
+    // add the listener to dispatch messages later
+    connection.listener(new ExtendedListener() {
+      public void onConnected() {
+        logger.debug("Connected");
+      }
+
+      @Override
+      public void onDisconnected() {
+        logger.debug("Disconnected");
+      }
+
+      @Override
+      public void onPublish(UTF8Buffer topic, Buffer body, Callback<Callback<Void>> ack) {
+        String topicString = topic.toString();
+        Consumer<byte[]> callback = callbacks.get(topicString);
+        if (callback == null) {
+          logger.debug("Got a message, but no callback to call. Forgot to unsubscribe?");
+        } else {
+          byte[] message = body.toByteArray();
+//          System.out.println("message = " + Arrays.toString(message));
+          callback.accept(message);
+        }
+        ack.onSuccess(null);  // always acknowledge message
+      }
+
+      @Override
+      public void onPublish(UTF8Buffer topicBuffer, Buffer body, Runnable ack) {
+        logger.warn("onPublish should not be called");
+      }
+
+      @Override
+      public void onFailure(Throwable cause) {
+//        logger.catching(cause);
+        error.set(cause);
+      }
+    });
+    throwIf(error);
+
+    // actually establish the connection
+    connection.connect(new Callback<Void>() {
+      @Override
+      public void onSuccess(Void value) {
+        connection.publish("components", (name + " is connected").getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
+          @Override
+          public void onSuccess(Void value) {
+            logger.debug("success sending welcome message");
+            try {
+              readyLock.lock();
+              ready = true;
+              readyCondition.signalAll();
+            } finally {
+              readyLock.unlock();
+            }
+          }
+
+          @Override
+          public void onFailure(Throwable value) {
+            logger.debug("failure sending welcome message", value);
+          }
+        });
+      }
+
+      @Override
+      public void onFailure(Throwable cause) {
+//        logger.error("Could not connect", cause);
+        error.set(cause);
+      }
+    });
+    throwIf(error);
+    return this;
+  }
+
+  public URI getHost() {
+    return host;
+  }
+
+  private void throwIf(AtomicReference<Throwable> error) throws IOException {
+    if (error.get() != null) {
+      throw new IOException(error.get());
+    }
+  }
+
+  public void setQoSForSubscription(QoS qos) {
+    this.qos = qos;
+  }
+
+  public void newConnection(String topic, Consumer<byte[]> callback) {
+    if (!ready) {
+      // TODO should maybe be something more kind than throwing an exception here
+      throw new IllegalStateException("Updater not ready");
+    }
+    // register callback
+    callbacks.put(topic, callback);
+
+    // subscribe at broker
+    Topic[] topicArray = { new Topic(topic, this.qos) };
+    connection.subscribe(topicArray, new Callback<byte[]>() {
+      @Override
+      public void onSuccess(byte[] qoses) {
+        logger.debug("Subscribed to {}, qoses: {}", topic, qoses);
+      }
+
+      @Override
+      public void onFailure(Throwable cause) {
+        logger.error("Could not subscribe to {}", topic, cause);
+      }
+    });
+  }
+
+  /**
+   * Waits until this updater is ready to receive MQTT messages.
+   * If it already is ready, return immediately with the value <code>true</code>.
+   * Otherwise waits for the given amount of time, and either return <code>true</code> within the timespan,
+   * if it got ready, or <code>false</code> upon a timeout.
+   * @param time the maximum time to wait
+   * @param unit the time unit of the time argument
+   * @return whether this updater is ready
+   */
+  public boolean waitUntilReady(long time, TimeUnit unit) {
+    try {
+      readyLock.lock();
+      if (ready) {
+        return true;
+      }
+      return readyCondition.await(time, unit);
+    } catch (InterruptedException e) {
+      e.printStackTrace();
+    } finally {
+      readyLock.unlock();
+    }
+    return false;
+  }
+
+  public void close() {
+    if (connection == null) {
+      logger.warn("Stopping without connection. Was setHost() called?");
+      return;
+    }
+    connection.disconnect(new Callback<Void>() {
+      @Override
+      public void onSuccess(Void value) {
+        logger.info("Disconnected {} from {}", name, host);
+      }
+
+      @Override
+      public void onFailure(Throwable ignored) {
+        // Disconnects never fail. And we do not care either.
+      }
+    });
+  }
+
+  public void publish(String topic, byte[] bytes) {
+    connection.publish(topic, bytes, qos, false, new Callback<Void>() {
+      @Override
+      public void onSuccess(Void value) {
+        logger.debug("Published some bytes to {}", topic);
+      }
+
+      @Override
+      public void onFailure(Throwable value) {
+        logger.warn("Could not publish on topic '{}'", topic);
+      }
+    });
+  }
+}

src/main/jastadd/NameResolution.jrag

@@ -27,7 +27,7 @@ aspect NameResolution {
     return null;
   }
 
-  refine RefResolverStubs eq UpdateDefinition.resolveMappingByToken(String id) {
+  refine RefResolverStubs eq UpdateDefinition.resolveMappingByToken(String id, int position) {
     // return a MappingDefinition
     for (MappingDefinition mappingDefinition : ros2rag().getMappingDefinitionList()) {
       if (mappingDefinition.getID().equals(id)) {

src/main/jastadd/Ros2Rag.flex

@@ -59,7 +59,6 @@ ID = [a-zA-Z$_][a-zA-Z0-9$_]*
 "maps"       { return sym(Terminals.MAPS); }
 "to"         { return sym(Terminals.TO); }
 "as"         { return sym(Terminals.AS); }
-"with"       { return sym(Terminals.WITH); }
 
 ";"          { return sym(Terminals.SCOL); }
 ":"          { return sym(Terminals.COL); }

src/main/jastadd/Ros2Rag.parser

@@ -6,6 +6,13 @@ Ros2Rag ros2rag
   |                                     {: return new Ros2Rag(); :}
 ;
 
+%embed {:
+  private Iterable<String> makeMappingDefs(ArrayList<?> raw_mapping_defs) {
+    java.util.Collections.reverse(raw_mapping_defs);
+    return () -> raw_mapping_defs.stream().map(raw -> ((Symbol) raw).value.toString()).iterator();
+  }
+:} ;
+
 // read Joint.CurrentPosition using LinkStateToIntPosition ;
 // write RobotArm._AppropriateSpeed using CreateSpeedMessage ;
 UpdateDefinition update_definition
@@ -15,11 +22,13 @@ UpdateDefinition update_definition
       result.setToken(TokenComponent.createRef(type_name + "." + token_name));
       return result;
     :}
-  | READ ID.type_name DOT ID.token_name USING ID.mapping_def SCOL
+  | READ ID.type_name DOT ID.token_name USING string_list.mapping_defs SCOL
     {:
       ReadFromMqttDefinition result = new ReadFromMqttDefinition();
       result.setToken(TokenComponent.createRef(type_name + "." + token_name));
-      result.setMapping(MappingDefinition.createRef(mapping_def));
+      for (String mapping_def : makeMappingDefs(mapping_defs)) {
+        result.addMapping(MappingDefinition.createRef(mapping_def));
+      }
       return result;
     :}
   | WRITE ID.type_name DOT ID.token_name SCOL
@@ -28,15 +37,22 @@ UpdateDefinition update_definition
       result.setToken(TokenComponent.createRef(type_name + "." + token_name));
       return result;
     :}
-  | WRITE ID.type_name DOT ID.token_name USING ID.mapping_def SCOL
+  | WRITE ID.type_name DOT ID.token_name USING string_list.mapping_defs SCOL
     {:
       WriteToMqttDefinition result = new WriteToMqttDefinition();
       result.setToken(TokenComponent.createRef(type_name + "." + token_name));
-      result.setMapping(MappingDefinition.createRef(mapping_def));
+      for (String mapping_def : makeMappingDefs(mapping_defs)) {
+        result.addMapping(MappingDefinition.createRef(mapping_def));
+      }
       return result;
     :}
 ;
 
+ArrayList string_list
+  = ID
+  | string_list COMMA ID
+;
+
 // RobotArm._AppropriateSpeed canDependOn Joint.CurrentPosition as dependency1 ;
 DependencyDefinition dependency_definition
   = ID.target_type DOT ID.target_token CAN_DEPEND_ON ID.source_type DOT ID.source_token AS ID.id SCOL
@@ -54,43 +70,29 @@ DependencyDefinition dependency_definition
 //  y = IntPosition.of((int) p.getPositionX(), (int) p.getPositionY(), (int) p.getPositionZ());
 //}
 MappingDefinition mapping_definition
-  = ID.id MAPS mapping_type.from TO mapping_type.to MAPPING_CONTENT.content
+  = ID.id MAPS mapping_type.from_type ID.from_name TO mapping_type.to_type MAPPING_CONTENT.content
     {:
       MappingDefinition result = new MappingDefinition();
       result.setID(id);
-      result.setFrom(from);
-      result.setTo(to);
+      result.setFromType(from_type);
+      result.setFromVariableName(from_name);
+      result.setToType(to_type);
       result.setContent(content.substring(2, content.length() - 2));
       return result;
     :}
 ;
 
 MappingDefinitionType mapping_type
-  = java_type_use.type ID.variable
+  = java_type_use.type
     {:
-      MappingDefinitionType result = new MappingDefinitionType();
+      JavaMappingDefinitionType result = new JavaMappingDefinitionType();
       result.setType(type);
-      result.setVariableName(variable);
       return result;
     :}
-  | java_type_use.type ID.variable WITH ID.method
+  | java_type_use.type LBRACKET RBRACKET
     {:
-      MappingDefinitionType result = new MappingDefinitionType();
+      JavaArrayMappingDefinitionType result = new JavaArrayMappingDefinitionType();
       result.setType(type);
-      result.setVariableName(variable);
-      result.setSerializationMethodName(method);
       return result;
     :}
 ;
-
-//String mapping_def_content
-//  = LB_CURLY COL mapping_def_content_body.b   {: return b.stream().collect(java.util.stream.Collectors.joining("\n")); :}
-//;
-//
-//ArrayList mapping_def_content_body
-//  = COL RB_CURLY                          {: return new ArrayList(); :}
-//  | TEXT.text mapping_def_content_body.b  {: b.add(0, text); return b; :}
-//;
-//String mapping_def_content
-//  = MappingContent.c      {: int length = c.length(); return c.substring(2, length - 2); :}
-//;

src/main/jastadd/Ros2Rag.relast

@@ -2,7 +2,7 @@ Ros2Rag ::= UpdateDefinition* DependencyDefinition* MappingDefinition* Program;
 
 abstract UpdateDefinition ::= <AlwaysApply:Boolean> ;
 
-rel UpdateDefinition.Mapping? -> MappingDefinition;
+rel UpdateDefinition.Mapping* -> MappingDefinition;
 
 abstract TokenUpdateDefinition : UpdateDefinition;
 rel TokenUpdateDefinition.Token -> TokenComponent;
@@ -15,5 +15,7 @@ DependencyDefinition ::= <ID> ;
 rel DependencyDefinition.Source -> TokenComponent ;
 rel DependencyDefinition.Target -> TokenComponent ;
 
-MappingDefinition ::= <ID> From:MappingDefinitionType To:MappingDefinitionType <Content> ;
-MappingDefinitionType ::= Type:JavaTypeUse <VariableName> <SerializationMethodName> ;  // SerializationMethodName may be empty
+MappingDefinition ::= <ID> FromType:MappingDefinitionType <FromVariableName> ToType:MappingDefinitionType <Content> ;
+abstract MappingDefinitionType ::= ;
+JavaMappingDefinitionType : MappingDefinitionType ::= Type:JavaTypeUse ;
+JavaArrayMappingDefinitionType : MappingDefinitionType ::= Type:JavaTypeUse ;

src/main/jastadd/backend/Aspect.jadd

@@ -18,18 +18,23 @@ aspect Aspect {
     sb.append("}\n");
   }
 
-  public String Ros2Rag.generateAspect() {
+  public String Ros2Rag.generateAspect(String rootNodeName) {
     StringBuilder sb = new StringBuilder();
-    generateAspect(sb);
+    generateMqttAspect(sb);
+    generateGrammarExtension(sb);
     return sb.toString();
   }
 
+  public void Ros2Rag.generateMqttAspect(StringBuilder sb) {
+
+  }
+
   // from "[always] read Joint.CurrentPosition using PoseToPosition;" generate method connectTo
 //    Joint j;
 //    j.getCurrentPosition().connectTo("/robot/joint2/pos");
 
-  public void Ros2Rag.generateAspect(StringBuilder sb) {
-    sb.append("aspect ROS2RAG {\n");
+  public void Ros2Rag.generateGrammarExtension(StringBuilder sb) {
+    sb.append("aspect ros2rag.GrammarExtension {\n");
 
     for (UpdateDefinition def : getUpdateDefinitionList()) {
       def.generateAspect(sb);

src/main/java/org/jastadd/ros2rag/compiler/Compiler.java

@@ -19,6 +19,7 @@ public class Compiler {
 
   private StringOption optionOutputDir;
   private StringOption optionInputGrammar;
+  private StringOption optionRootNode;
   private StringOption optionInputRos2Rag;
 
   private ArrayList<Option<?>> options;
@@ -44,12 +45,8 @@ public class Compiler {
 
     printMessage("Running ROS2RAG Preprocessor");
 
-    if (!optionInputGrammar.isSet()) {
-      return error("specify an input grammar");
-    }
-
-    if (!optionInputRos2Rag.isSet()) {
-      return error("specify the ros2rag definition file");
+    if (anyRequiredOptionIsUnset()) {
+      return error("Aborting.");
     }
 
     List<String> otherArgs = commandLine.getArguments();
@@ -59,11 +56,30 @@ public class Compiler {
     Ros2Rag ros2Rag = parseProgram(optionInputGrammar.getValue(), optionInputRos2Rag.getValue());
 
     printMessage("Writing output files");
+    // copy MqttUpdater into outputDir
+    try {
+      Files.copy(Paths.get("src", "main", "jastadd", "MqttUpdater.java_class"),
+          Paths.get(outputDir, "MqttUpdater.java"));
+    } catch (IOException e) {
+      throw new CompilerException("Could not copy MqttUpdater.java", e);
+    }
     writeToFile(outputDir + "/Grammar.relast", ros2Rag.getProgram().generateAbstractGrammar());
-    writeToFile(outputDir + "/ROS2RAG.jadd", ros2Rag.generateAspect());
+    writeToFile(outputDir + "/ROS2RAG.jadd", ros2Rag.generateAspect(optionRootNode.getValue()));
     return 0;
   }
 
+  private boolean anyRequiredOptionIsUnset() {
+    boolean foundError = false;
+    for (Option<?> option : options) {
+      if (option.hasArgument() == Option.HasArgument.YES && !option.isSet()) {
+        System.err.println("Option '" + option.getName() +
+            "' (" + option.getDescription() + ") is required but unset!");
+        foundError = true;
+      }
+    }
+    return foundError;
+  }
+
 
   public static int main(String[] args) {
     try {
@@ -91,6 +107,7 @@ public class Compiler {
   private void addOptions() {
     optionOutputDir = addOption(new StringOption("outputDir", "target directory for the generated files."));
     optionInputGrammar = addOption(new StringOption("inputGrammar", "base grammar."));
+    optionRootNode = addOption(new StringOption("rootNode", "root node in the base grammar."));
     optionInputRos2Rag = addOption(new StringOption("inputRos2Rag", "ros2rag definition file."));
   }
 

src/main/java/org/jastadd/ros2rag/compiler/SimpleMain.java

@@ -51,8 +51,9 @@ public class SimpleMain {
 
     MappingDefinition mappingDefinition = new MappingDefinition();
     mappingDefinition.setID("PoseToPosition");
-    mappingDefinition.setFrom(makeMappingDefinitionType("PBPose", "x"));
-    mappingDefinition.setTo(makeMappingDefinitionType("Position", "y"));
+    mappingDefinition.setFromType(makeMappingDefinitionType("PBPose"));
+    mappingDefinition.setFromVariableName("x");
+    mappingDefinition.setToType(makeMappingDefinitionType("Position"));
     mappingDefinition.setContent("      pose.position.x += sqrt(.5 * size.x)\n" +
         "      MAP round(2)\n" +
         "      x = x / 100\n" +
@@ -63,18 +64,17 @@ public class SimpleMain {
     ReadFromMqttDefinition readFromMqttDefinition = new ReadFromMqttDefinition();
     readFromMqttDefinition.setAlwaysApply(false);
     readFromMqttDefinition.setToken(TokenComponent.createRef("Joint.CurrentPosition"));
-    readFromMqttDefinition.setMapping(mappingDefinition);
+    readFromMqttDefinition.addMapping(mappingDefinition);
     model.addUpdateDefinition(readFromMqttDefinition);
 
     model.treeResolveAll();
 
-    System.out.println(model.generateAspect());
+    System.out.println(model.generateAspect("Model"));
   }
 
-  private static MappingDefinitionType makeMappingDefinitionType(String type, String variable) {
-    MappingDefinitionType result = new MappingDefinitionType();
+  private static MappingDefinitionType makeMappingDefinitionType(String type) {
+    JavaMappingDefinitionType result = new JavaMappingDefinitionType();
     result.setType(new SimpleJavaTypeUse(type));
-    result.setVariableName(variable);
     return result;
   }
 

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

@@ -12,7 +12,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class RosToRagTest {
 
-  void transform(String inputGrammar, String inputRos2Rag, String outputDir) throws CommandLine.CommandLineException, Compiler.CompilerException {
+  void transform(String inputGrammar, String inputRos2Rag, String rootNode, String outputDir) throws CommandLine.CommandLineException, Compiler.CompilerException {
 
     System.out.println(Paths.get(".").toAbsolutePath());
     assertTrue(Paths.get(inputGrammar).toFile().exists(), "input grammar does not exist");
@@ -30,7 +30,8 @@ public class RosToRagTest {
     String[] args = {
         "--outputDir=" + outputDir,
         "--inputGrammar=" + inputGrammar,
-        "--inputRos2Rag=" + inputRos2Rag
+        "--inputRos2Rag=" + inputRos2Rag,
+        "--rootNode=" + rootNode
     };
 
     new Compiler().run(args);
@@ -40,6 +41,7 @@ public class RosToRagTest {
   void transformMinimalExample() throws CommandLine.CommandLineException, Compiler.CompilerException {
     transform("src/test/resources/Example.relast",
         "src/test/resources/Example.ros2rag",
+        "Model",
         "src/test/resources/out");
   }
 }

src/test/resources/Example.ros2rag

 /* Version 2020-04-17 */
 // --- update definitions ---
-read Joint.CurrentPosition using LinkStateToIntPosition ;
-write RobotArm._AppropriateSpeed using CreateSpeedMessage ;
+read Joint.CurrentPosition using ParseLinkState, LinkStateToIntPosition ;
+write RobotArm._AppropriateSpeed using CreateSpeedMessage, SerializeRobotConfig ;
 
 // --- dependency definitions ---
 RobotArm._AppropriateSpeed canDependOn Joint.CurrentPosition as dependency1 ;
 RobotArm._AppropriateSpeed canDependOn RobotArm._AttributeTestSource as dependency2 ;
 
 // --- mapping definitions ---
-LinkStateToIntPosition maps panda.Linkstate.PandaLinkState x to IntPosition y {:
-  panda.Linkstate.PandaLinkState.Position p = x.getPos();
+ParseLinkState maps byte[] bytes to panda.Linkstate.PandaLinkState {:
+  return panda.Linkstate.PandaLinkState.parseFrom(bytes);
+:}
+
+SerializeRobotConfig maps config.Robotconfig.RobotConfig rc to byte[] {:
+  return rc.toByteArray();
+:}
+
+LinkStateToIntPosition maps panda.Linkstate.PandaLinkState pls to IntPosition {:
+  panda.Linkstate.PandaLinkState.Position p = pls.getPos();
   { int i = 0; }
-  y = IntPosition.of((int) p.getPositionX(), (int) p.getPositionY(), (int) p.getPositionZ());
+  return IntPosition.of((int) p.getPositionX(), (int) p.getPositionY(), (int) p.getPositionZ());
 :}
 
-CreateSpeedMessage maps double x to config.Robotconfig.RobotConfig y {:
-  y = config.Robotconfig.RobotConfig.newBuilder()
-    .setSpeed(x)
+CreateSpeedMessage maps double speed to config.Robotconfig.RobotConfig {:
+  return config.Robotconfig.RobotConfig.newBuilder()
+    .setSpeed(speed)
     .build();
 :}