Add realtime inference consumer

build.gradle

@@ -37,13 +37,14 @@ ext {
     findbugVersion = '3.0.2'
     commonsCliVersion = '1.4'
     mockitoVersion = '3.5.11'
-    radarCommonsVersion = '0.13.0'
-    radarSchemasVersion = '0.5.14'
+    radarCommonsVersion = '0.13.2'
+    radarSchemasVersion = '0.7.4'
     subethamailVersion = '3.1.7'
     jsoupVersion = '1.13.1'
     slf4jVersion = '1.7.30'
     log4jVersion = '1.2.17'
     avroVersion = '1.9.2'
+    jsonPathVersion = '2.6.0'
 }
 
 //---------------------------------------------------------------------------//
@@ -53,9 +54,6 @@ ext {
 // In this section you declare where to find the dependencies of your project
 repositories {
     mavenCentral()
-    // Non-jcenter radar releases
-    maven { url 'http://dl.bintray.com/radar-cns/org.radarcns' }
-    maven { url  'http://dl.bintray.com/radar-base/org.radarbase' }
     // Kafka/confluent releases
     maven { url 'http://packages.confluent.io/maven/' }
     // For working with dev-branches
@@ -69,7 +67,9 @@ dependencies {
     implementation group: 'org.radarbase', name: 'radar-commons', version: radarCommonsVersion
     implementation group: 'org.apache.avro', name: 'avro', version: avroVersion
     implementation group: 'org.radarbase', name: 'radar-commons-testing', version: radarCommonsVersion
-    implementation group: 'org.radarcns', name: 'radar-schemas-commons', version: radarSchemasVersion
+    implementation group: 'org.radarbase', name: 'radar-schemas-commons', version:
+            radarSchemasVersion
+    implementation group: 'org.radarbase', name: 'oauth-client-util', version: '0.8.0'
 
     // Kafka streaming API
     implementation group: 'org.apache.kafka', name: 'kafka-streams', version: kafkaVersion
@@ -92,6 +92,9 @@ dependencies {
     // JSoup html parser
     implementation group: 'org.jsoup', name: 'jsoup', version: jsoupVersion
 
+    // JsonPath for evaluating json conditions dynamically.
+    implementation group: 'com.jayway.jsonpath', name: 'json-path', version: jsonPathVersion
+
     runtimeOnly group: 'org.radarbase', name: 'radar-commons-unsafe', version: radarCommonsVersion
 
     runtimeOnly group: 'log4j', name: 'log4j', version: log4jVersion

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,4 @@
-distributionBase=GRADLE_USER_HOME
-distributionPath=wrapper/dists
+#Fri Oct 04 14:28:19 BST 2019
 distributionUrl=https\://services.gradle.org/distributions/gradle-6.6.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

src/main/java/org/radarcns/RadarBackend.java

@@ -23,6 +23,7 @@
 import org.radarcns.config.RadarBackendOptions;
 import org.radarcns.config.RadarPropertyHandler;
 import org.radarcns.config.SubCommand;
+import org.radarcns.consumer.realtime.RealtimeInferenceConsumerFactory;
 import org.radarcns.monitor.KafkaMonitorFactory;
 import org.radarcns.producer.MockProducerCommand;
 import org.radarcns.stream.KafkaStreamFactory;
@@ -76,6 +77,8 @@ public SubCommand createCommand() throws IOException {
                 return new KafkaMonitorFactory(options, radarPropertyHandler).createMonitor();
             case "mock":
                 return new MockProducerCommand(options, radarPropertyHandler);
+            case "realtime_consumers":
+                return RealtimeInferenceConsumerFactory.createConsumersFor(radarPropertyHandler);
             default:
                 throw new IllegalArgumentException("Unknown subcommand "
                         + options.getSubCommand());

src/main/java/org/radarcns/config/ConfigRadar.java

@@ -23,6 +23,7 @@
 
 import org.radarbase.config.ServerConfig;
 import org.radarbase.config.YamlConfigLoader;
+import org.radarcns.config.realtime.RealtimeConsumerConfig;
 
 /**
  * POJO representing the yml file
@@ -42,6 +43,8 @@ public class ConfigRadar {
     private DisconnectMonitorConfig disconnectMonitor;
     @JsonProperty("statistics_monitors")
     private List<SourceStatisticsStreamConfig> statisticsMonitors;
+    @JsonProperty("realtime_consumers")
+    private List<RealtimeConsumerConfig> consumerConfigs;
     @JsonProperty("stream")
     private StreamConfig stream;
     @JsonProperty("persistence_path")
@@ -182,6 +185,14 @@ public void setStatisticsMonitors(List<SourceStatisticsStreamConfig> statisticsM
         this.statisticsMonitors = statisticsMonitors;
     }
 
+    public List<RealtimeConsumerConfig> getConsumerConfigs() {
+        return consumerConfigs;
+    }
+
+    public void setConsumerConfigs(List<RealtimeConsumerConfig> consumerConfigs) {
+        this.consumerConfigs = consumerConfigs;
+    }
+
     public StreamConfig getStream() {
         return stream;
     }

src/main/java/org/radarcns/config/realtime/ActionConfig.java

@@ -0,0 +1,24 @@
+package org.radarcns.config.realtime;
+
+import java.util.Map;
+
+public class ActionConfig {
+    private String name;
+    private Map<String, Object> properties;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Map<String, Object> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, Object> properties) {
+        this.properties = properties;
+    }
+}

src/main/java/org/radarcns/config/realtime/ConditionConfig.java

@@ -0,0 +1,24 @@
+package org.radarcns.config.realtime;
+
+import java.util.Map;
+
+public class ConditionConfig {
+    private String name;
+    private Map<String, Object> properties;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Map<String, Object> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, Object> properties) {
+        this.properties = properties;
+    }
+}

src/main/java/org/radarcns/config/realtime/RealtimeConsumerConfig.java

@@ -0,0 +1,61 @@
+package org.radarcns.config.realtime;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import java.util.Map;
+
+public class RealtimeConsumerConfig {
+
+    private String name;
+
+    private String topic;
+
+    @JsonProperty("conditions")
+    private List<ConditionConfig> conditionConfigs;
+
+    @JsonProperty("actions")
+    private List<ActionConfig> actionConfigs;
+
+    @JsonProperty("consumer_properties")
+    private Map<String, String> consumerProperties;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public List<ConditionConfig> getConditionConfigs() {
+        return conditionConfigs;
+    }
+
+    public void setConditionConfigs(List<ConditionConfig> conditionConfigs) {
+        this.conditionConfigs = conditionConfigs;
+    }
+
+    public List<ActionConfig> getActionConfigs() {
+        return actionConfigs;
+    }
+
+    public void setActionConfigs(List<ActionConfig> actionConfigs) {
+        this.actionConfigs = actionConfigs;
+    }
+
+    public Map<String, String> getConsumerProperties() {
+        return consumerProperties;
+    }
+
+    public void setConsumerProperties(Map<String, String> consumerProperties) {
+        this.consumerProperties = consumerProperties;
+    }
+}