xAPI client integration

.github/workflows/ci.yml

@@ -22,4 +22,4 @@ jobs:
                   cache: maven
 
             - name: Verify
-              run: mvn --batch-mode verify
+              run: make verify

Makefile

@@ -30,8 +30,8 @@ lint:
 test:
 	mvn test
 
-verify:
-	mvn verify
+verify: lib
+	mvn --batch-mode verify
 
 run-dev:
 	java -cp "$(APP_JAR):$(PORTICO_JAR)" com.yetanalytics.hlaxapi.App $(SIM_CONFIG)

README.md

@@ -45,6 +45,12 @@ Example:
 }
 ```
 
+The frequency of the buffer being cleared and statement being actually posted can be changed with a java property when running the jar:
+
+```
+java -Dxapi.buffer.clear-rate=5000 ...
+```
+
 ### Vendoring Portico
 
 `make lib` rebuilds and vendors only Portico's Java jar into this repository's local Maven file repository. Portico ships its own Ant wrapper at `codebase/ant`, so no separate Ant installation is required.

config/xapi-config.json

@@ -20,14 +20,21 @@
                 },
                 "result": {
                     "response": "[<<[\"this\", [\"ToPosition\", \"X\"]]>>, <<[\"this\", [\"ToPosition\", \"Y\"]]>>]"
+                },
+                "verb": {
+                    "id": "http://yetanalytics.com/verbs/moved"
+                },
+                "object": {
+                    "id": "http://yetanalytics.com/objects/7"
                 }
             }
         }
     ],
     "lrs": {
-        "host": "host string",
-        "key": "key string",
-        "secret": "secret string",
-        "batch": 4
+        "host": "http://localhost:8080/xapi",
+        "key": "my_key",
+        "secret": "my_secret",
+        "batch": 35,
+        "maxRetries": 3
     }    
 }

pom.xml

@@ -105,6 +105,12 @@
             <artifactId>spring-context</artifactId>
             <version>6.1.18</version>
         </dependency>
+        <!-- xAPI Client -->
+        <dependency>
+            <groupId>com.yetanalytics</groupId>
+            <artifactId>xapi-tools</artifactId>
+            <version>0.0.4</version>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/yetanalytics/hlaxapi/AppConfig.java

@@ -10,6 +10,7 @@
 import org.apache.logging.log4j.Logger;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 import com.yetanalytics.hlaxapi.config.ConfigParser;
 import com.yetanalytics.hlaxapi.config.XapiConfig;
@@ -21,6 +22,7 @@
  * Spring configuration class for beans related to HLA and simulation configuration.
  */
 @Configuration
+@EnableScheduling
 public class AppConfig {
 
     private static final Logger logger = LogManager.getLogger(AppConfig.class);

src/main/java/com/yetanalytics/hlaxapi/HlaInterfaceImpl.java

@@ -9,6 +9,8 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
 
 import com.yetanalytics.hlaxapi.config.XapiConfig;
 import com.yetanalytics.hlaxapi.config.model.StatementTrigger;
@@ -57,9 +59,6 @@
 import hla.rti1516e.exceptions.SaveInProgress;
 import hla.rti1516e.exceptions.UnsupportedCallbackModel;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
 @Component
 public class HlaInterfaceImpl extends NullFederateAmbassador implements HlaInterface {
 
@@ -78,6 +77,9 @@ public class HlaInterfaceImpl extends NullFederateAmbassador implements HlaInter
     @Autowired
     private TriggerProcessor triggerProcessor;
 
+    @Autowired
+    private XapiClient xapiClient;
+
     public void start()
             throws ConnectionFailed, InvalidLocalSettingsDesignator, RTIinternalError, NotConnected, ErrorReadingFDD,
             CouldNotOpenFDD, InconsistentFDD, RestoreInProgress, SaveInProgress,
@@ -213,7 +215,7 @@ private void receiveInteraction(InteractionClassHandle interactionClass, Paramet
         logger.info("Received Interaction");
         try {
             String interactionName = ambassador.getInteractionClassName(interactionClass);
-            logger.info("Interaction Handle: {}", interactionName);
+            logger.trace("Interaction Handle: {}", interactionName);
             String interactionKey = StringUtils.substringAfterLast(interactionName, ".");
 
             // Create Interaction-specific injection context to pass to trigger processor
@@ -225,8 +227,13 @@ private void receiveInteraction(InteractionClassHandle interactionClass, Paramet
                     .filter(trigger -> trigger.clazz.equals(interactionKey)
                             && trigger.type.equals(StatementTrigger.Type.INTERACTION))
                     .forEach(trigger -> {
-                        logger.info("Processing trigger for interaction {}", trigger.clazz);
-                        triggerProcessor.processTrigger(trigger, context);
+                        logger.trace("Processing trigger for interaction {}", trigger.clazz);
+                        String xapi = triggerProcessor.processTrigger(trigger, context);
+                        try {
+                            xapiClient.sendStatementFromString(xapi);
+                        } catch (Exception e) {
+                            logger.error("Error parsing or posting statement {}", xapi, e);
+                        }
                     });
         } catch (InvalidInteractionClassHandle | FederateNotExecutionMember | NotConnected | RTIinternalError e) {
             logger.error("Error ascertaining interaction details!", e);

src/main/java/com/yetanalytics/hlaxapi/TriggerProcessor.java

@@ -48,7 +48,7 @@ public String processTrigger(StatementTrigger trigger, InjectionContext context)
             JsonNode processed = processNode(stmtNode, context, mapper);
 
             String output = mapper.writeValueAsString(processed);
-            logger.info("Processed statement output: {}", output);
+            logger.trace("Processed statement output: {}", output);
             return output;
         } catch (Exception e) {
             logger.debug("Could not process statement for trigger", e);

src/main/java/com/yetanalytics/hlaxapi/XapiClient.java

@@ -0,0 +1,98 @@
+package com.yetanalytics.hlaxapi;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.yetanalytics.hlaxapi.config.XapiConfig;
+
+import com.yetanalytics.xapi.client.StatementClient;
+import com.yetanalytics.xapi.exception.StatementClientException;
+import com.yetanalytics.xapi.model.Statement;
+import com.yetanalytics.xapi.client.LRS;
+
+import com.yetanalytics.xapi.util.Mapper;
+
+@Component
+public class XapiClient {
+
+    private static final Logger logger = LogManager.getLogger(XapiClient.class);
+
+    private StatementClient client;
+
+    private List<Statement> buffer;
+
+    private Integer batchSize;
+
+    private Integer retryCount = 0;
+
+    private Integer maxRetries;
+
+    public XapiClient(XapiConfig xapiConfig) {
+        LRS lrs = new LRS(
+            xapiConfig.lrsConfig.host,
+            xapiConfig.lrsConfig.key,
+            xapiConfig.lrsConfig.secret,
+            xapiConfig.lrsConfig.batch
+        );
+        client = new StatementClient(lrs);
+        buffer = new ArrayList<Statement>();
+        batchSize = xapiConfig.lrsConfig.batch;
+        maxRetries = xapiConfig.lrsConfig.maxRetries;
+    }
+
+    private StatementClient getClient() {
+        return client;
+    }
+
+    public void sendStatementFromString(String s) throws JsonMappingException, JsonProcessingException{
+        Statement stmt = Mapper.getMapper().readValue(s, Statement.class);
+        addToBuffer(stmt);
+    }
+
+    /** Synchronized buffer methods */
+
+    private synchronized void addToBuffer(Statement stmt){
+        buffer.add(stmt);
+    }
+
+    // Check and post buffer to LRS every 10 seconds (or ENV) if contains statements
+    @Scheduled(fixedRateString = "${xapi.buffer.clear-rate:10000}")
+    private synchronized void clearBuffer() {
+        logger.info("Buffer Size: {}", buffer.size());
+        if (buffer.size() > 0){
+            try {
+                List<UUID> results = client.postStatements(buffer);
+                logger.info("Stored statements: {}", results);
+                buffer = new ArrayList<Statement>();
+                logger.info("Cleared Buffer");
+                retryCount = 0;
+            } catch (StatementClientException e) {
+                logger.error("Error sending statements to LRS:", e);
+                if (retryCount < maxRetries) {
+                    retryCount++;
+                    logger.info("Retrying to send statements to LRS, attempt {}/{}", retryCount, maxRetries);
+                } else {
+                    // TODO: For durability, we should write the buffer to a file or database or DLQ for retrying later
+                    // might be a case for a light queueing system like RabbitMQ. Also failures will be reduced if we
+                    // pre-validate statements before sending to the LRS. Also we may want to reduce the accrual of
+                    // statements in the buffer if we have a DLQ strategy, that way less innocent statements are lost.
+                    // For now, we will just clear the buffer after max retries.
+                    // TODO: Additionally we should consider a cap on the buffer size, and if it exceeds that cap,
+                    // we should start dropping statements or writing to a DLQ.
+                    buffer = new ArrayList<Statement>();
+                    retryCount = 0;
+                    logger.info("Cleared Buffer after max retries, statement data lost!");
+                }
+            }
+        }
+    }
+
+}

src/main/java/com/yetanalytics/hlaxapi/config/model/LrsConfig.java

@@ -5,9 +5,10 @@ public class LrsConfig {
     public String key;
     public String secret;
     public int batch;
+    public int maxRetries;
 
     @Override
     public String toString() {
-        return String.format("LrsConfig{host=%s,key=%s,batch=%d}", host, key, batch);
+        return String.format("LrsConfig{host=%s,key=%s,batch=%d,maxRetries=%d}", host, key, batch, maxRetries);
     }
 }