Use latest RADAR-Schemas

build.gradle

@@ -54,7 +54,7 @@ allprojects {
     ext.mathVersion = '3.0'
     ext.hamcrestVersion = '1.3'
     ext.codacyVersion = '1.0.10'
-    ext.radarSchemasVersion = '0.1'
+    ext.radarSchemasVersion = '0.2-alpha.3-SNAPSHOT'
 
     ext.githubUrl = 'https://github.com/' + githubRepoName + '.git'
     ext.issueUrl = 'https://github.com/' + githubRepoName + '/issues'
@@ -67,6 +67,9 @@ allprojects {
         jcenter()
         maven { url 'http://packages.confluent.io/maven/' }
         maven { url 'http://dl.bintray.com/typesafe/maven-releases' }
+        flatDir {
+            dirs "${project.rootDir}/libs"
+        }
     }
 
     ext.pomConfig = {

libs/.gitignore

@@ -0,0 +1 @@
+*.jar

src/main/java/org/radarcns/topic/SensorTopic.java

@@ -37,6 +37,9 @@ public SensorTopic(String name, Schema keySchema, Schema valueSchema,
             throw new IllegalArgumentException("Sensors must send records as values");
         }
 
+        if (keySchema.getField("projectId") == null) {
+            throw new IllegalArgumentException("Key schema must have a project ID");
+        }
         if (keySchema.getField("userId") == null) {
             throw new IllegalArgumentException("Key schema must have a user ID");
         }

src/test/java/org/radarcns/data/SpecificRecordDecorderTest.java

@@ -20,8 +20,8 @@
 
 import java.io.IOException;
 import org.junit.Test;
-import org.radarcns.empatica.EmpaticaE4BloodVolumePulse;
-import org.radarcns.key.MeasurementKey;
+import org.radarcns.passive.empatica.EmpaticaE4BloodVolumePulse;
+import org.radarcns.kafka.ObservationKey;
 import org.radarcns.topic.AvroTopic;
 
 /**
@@ -30,13 +30,14 @@
 public class SpecificRecordDecorderTest {
 
     @Test
-    public void decordJson() throws IOException {
+    public void decodeJson() throws IOException {
         SpecificRecordDecoder decoder = new SpecificRecordDecoder(false);
-        AvroTopic<MeasurementKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", MeasurementKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), MeasurementKey.class, EmpaticaE4BloodVolumePulse.class);
-        AvroDecoder.AvroReader<MeasurementKey> keyDecoder = decoder.reader(topic.getKeySchema(), topic.getKeyClass());
+        AvroTopic<ObservationKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", ObservationKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), ObservationKey.class, EmpaticaE4BloodVolumePulse.class);
+        AvroDecoder.AvroReader<ObservationKey> keyDecoder = decoder.reader(topic.getKeySchema(), topic.getKeyClass());
         AvroDecoder.AvroReader<EmpaticaE4BloodVolumePulse> valueDecoder = decoder.reader(topic.getValueSchema(), topic.getValueClass());
 
-        MeasurementKey key = keyDecoder.decode("{\"userId\":\"a\",\"sourceId\":\"b\"}".getBytes());
+        ObservationKey key = keyDecoder.decode("{\"projectId\":{\"string\":\"test\"},\"userId\":\"a\",\"sourceId\":\"b\"}".getBytes());
+        assertEquals(key.get("projectId"), "test");
         assertEquals(key.get("userId"), "a");
         assertEquals(key.get("sourceId"), "b");
 
@@ -47,15 +48,20 @@ public void decordJson() throws IOException {
     }
 
     @Test
-    public void decordBinary() throws IOException {
+    public void decodeBinary() throws IOException {
 
         SpecificRecordDecoder decoder = new SpecificRecordDecoder(true);
-        AvroTopic<MeasurementKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", MeasurementKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), MeasurementKey.class, EmpaticaE4BloodVolumePulse.class);
-        AvroDecoder.AvroReader<MeasurementKey> keyDecoder = decoder.reader(topic.getKeySchema(), topic.getKeyClass());
+        AvroTopic<ObservationKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", ObservationKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), ObservationKey.class, EmpaticaE4BloodVolumePulse.class);
+        AvroDecoder.AvroReader<ObservationKey> keyDecoder = decoder.reader(topic.getKeySchema(), topic.getKeyClass());
         AvroDecoder.AvroReader<EmpaticaE4BloodVolumePulse> valueDecoder = decoder.reader(topic.getValueSchema(), topic.getValueClass());
 
-        byte[] inputKey = {2, 97, 2, 98};
-        MeasurementKey key = keyDecoder.decode( inputKey);
+        // note that positive numbers are multiplied by two in avro binary encoding, due to the
+        // zig-zag encoding schema used.
+        // See http://avro.apache.org/docs/1.8.1/spec.html#binary_encoding
+        // type index 1, length 4, char t, char e, char s, char t, length 1, char a, length 1, char b
+        byte[] inputKey = {2, 8, 116, 101, 115, 116, 2, 97, 2, 98};
+        ObservationKey key = keyDecoder.decode( inputKey);
+        assertEquals(key.get("projectId"), "test");
         assertEquals(key.get("userId"), "a");
         assertEquals(key.get("sourceId"), "b");
 

src/test/java/org/radarcns/data/SpecificRecordEncoderTest.java

@@ -18,8 +18,8 @@
 
 import junit.framework.TestCase;
 
-import org.radarcns.empatica.EmpaticaE4BloodVolumePulse;
-import org.radarcns.key.MeasurementKey;
+import org.radarcns.passive.empatica.EmpaticaE4BloodVolumePulse;
+import org.radarcns.kafka.ObservationKey;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -28,25 +28,28 @@
 public class SpecificRecordEncoderTest extends TestCase {
     public void testJson() throws IOException {
         SpecificRecordEncoder encoder = new SpecificRecordEncoder(false);
-        AvroTopic<MeasurementKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", MeasurementKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), MeasurementKey.class, EmpaticaE4BloodVolumePulse.class);
-        AvroEncoder.AvroWriter<MeasurementKey> keyEncoder = encoder.writer(topic.getKeySchema(), topic.getKeyClass());
+        AvroTopic<ObservationKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", ObservationKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), ObservationKey.class, EmpaticaE4BloodVolumePulse.class);
+        AvroEncoder.AvroWriter<ObservationKey> keyEncoder = encoder.writer(topic.getKeySchema(), topic.getKeyClass());
         AvroEncoder.AvroWriter<EmpaticaE4BloodVolumePulse> valueEncoder = encoder.writer(topic.getValueSchema(), topic.getValueClass());
 
-        byte[] key = keyEncoder.encode(new MeasurementKey("a", "b"));
+        byte[] key = keyEncoder.encode(new ObservationKey("test", "a", "b"));
         byte[] value = valueEncoder.encode(new EmpaticaE4BloodVolumePulse(0d, 0d, 0f));
-        assertEquals("{\"userId\":\"a\",\"sourceId\":\"b\"}", new String(key));
+        assertEquals("{\"projectId\":{\"string\":\"test\"},\"userId\":\"a\",\"sourceId\":\"b\"}", new String(key));
         assertEquals("{\"time\":0.0,\"timeReceived\":0.0,\"bloodVolumePulse\":0.0}", new String(value));
     }
 
     public void testBinary() throws IOException {
         SpecificRecordEncoder encoder = new SpecificRecordEncoder(true);
-        AvroTopic<MeasurementKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", MeasurementKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), MeasurementKey.class, EmpaticaE4BloodVolumePulse.class);
-        AvroEncoder.AvroWriter<MeasurementKey> keyEncoder = encoder.writer(topic.getKeySchema(), topic.getKeyClass());
+        AvroTopic<ObservationKey, EmpaticaE4BloodVolumePulse> topic = new AvroTopic<>("keeeeys", ObservationKey.getClassSchema(), EmpaticaE4BloodVolumePulse.getClassSchema(), ObservationKey.class, EmpaticaE4BloodVolumePulse.class);
+        AvroEncoder.AvroWriter<ObservationKey> keyEncoder = encoder.writer(topic.getKeySchema(), topic.getKeyClass());
         AvroEncoder.AvroWriter<EmpaticaE4BloodVolumePulse> valueEncoder = encoder.writer(topic.getValueSchema(), topic.getValueClass());
 
-        byte[] key = keyEncoder.encode(new MeasurementKey("a", "b"));
-        // start string, char a, start string, char b
-        byte[] expectedKey = {2, 97, 2, 98};
+        byte[] key = keyEncoder.encode(new ObservationKey("test", "a", "b"));
+        // note that positive numbers are multiplied by two in avro binary encoding, due to the
+        // zig-zag encoding schema used.
+        // See http://avro.apache.org/docs/1.8.1/spec.html#binary_encoding
+        // type index 1, length 4, char t, char e, char s, char t, length 1, char a, length 1, char b
+        byte[] expectedKey = {2, 8, 116, 101, 115, 116, 2, 97, 2, 98};
         System.out.println("key:      0x" + byteArrayToHex(key));
         System.out.println("expected: 0x" + byteArrayToHex(expectedKey));
         assertTrue(Arrays.equals(expectedKey, key));

src/test/java/org/radarcns/producer/rest/RestSenderTest.java

@@ -32,8 +32,8 @@
 import org.radarcns.config.ServerConfig;
 import org.radarcns.data.Record;
 import org.radarcns.data.SpecificRecordEncoder;
-import org.radarcns.key.MeasurementKey;
-import org.radarcns.phone.PhoneLight;
+import org.radarcns.kafka.ObservationKey;
+import org.radarcns.passive.phone.PhoneLight;
 import org.radarcns.producer.KafkaTopicSender;
 import org.radarcns.topic.AvroTopic;
 
@@ -47,7 +47,7 @@
 
 public class RestSenderTest {
     private SchemaRetriever retriever;
-    private RestSender<MeasurementKey, SpecificRecord> sender;
+    private RestSender<ObservationKey, SpecificRecord> sender;
 
     @Rule
     public MockWebServer webServer = new MockWebServer();
@@ -58,7 +58,7 @@ public void setUp() {
         SpecificRecordEncoder encoder = new SpecificRecordEncoder(false);
 
         ServerConfig config = new ServerConfig(webServer.url("/").url());
-        this.sender = new RestSender.Builder<MeasurementKey, SpecificRecord>()
+        this.sender = new RestSender.Builder<ObservationKey, SpecificRecord>()
                 .server(config)
                 .schemaRetriever(retriever)
                 .encoders(encoder, encoder)
@@ -68,18 +68,18 @@ public void setUp() {
 
     @Test
     public void sender() throws Exception {
-        Schema keySchema = MeasurementKey.getClassSchema();
+        Schema keySchema = ObservationKey.getClassSchema();
         Schema valueSchema = PhoneLight.getClassSchema();
-        AvroTopic<MeasurementKey, PhoneLight> topic = new AvroTopic<>("test",
-                keySchema, valueSchema, MeasurementKey.class, PhoneLight.class);
+        AvroTopic<ObservationKey, PhoneLight> topic = new AvroTopic<>("test",
+                keySchema, valueSchema, ObservationKey.class, PhoneLight.class);
         Headers headers = new Headers.Builder()
                 .add("Cookie", "ab")
                 .add("Cookie", "bc")
                 .build();
         sender.setHeaders(headers);
-        KafkaTopicSender<MeasurementKey, PhoneLight> topicSender = sender.sender(topic);
+        KafkaTopicSender<ObservationKey, PhoneLight> topicSender = sender.sender(topic);
 
-        MeasurementKey key = new MeasurementKey("a", "b");
+        ObservationKey key = new ObservationKey("test","a", "b");
         PhoneLight value = new PhoneLight(0.1, 0.2, 0.3f);
         ParsedSchemaMetadata keySchemaMetadata = new ParsedSchemaMetadata(10, 2, keySchema);
         ParsedSchemaMetadata valueSchemaMetadata = new ParsedSchemaMetadata(10, 2, valueSchema);
@@ -118,13 +118,13 @@ public void sender() throws Exception {
 
     @Test
     public void sendTwo() throws Exception {
-        Schema keySchema = MeasurementKey.getClassSchema();
+        Schema keySchema = ObservationKey.getClassSchema();
         Schema valueSchema = PhoneLight.getClassSchema();
-        AvroTopic<MeasurementKey, PhoneLight> topic = new AvroTopic<>("test",
-                keySchema, valueSchema, MeasurementKey.class, PhoneLight.class);
-        KafkaTopicSender<MeasurementKey, PhoneLight> topicSender = sender.sender(topic);
+        AvroTopic<ObservationKey, PhoneLight> topic = new AvroTopic<>("test",
+                keySchema, valueSchema, ObservationKey.class, PhoneLight.class);
+        KafkaTopicSender<ObservationKey, PhoneLight> topicSender = sender.sender(topic);
 
-        MeasurementKey key = new MeasurementKey("a", "b");
+        ObservationKey key = new ObservationKey("test", "a", "b");
         PhoneLight value = new PhoneLight(0.1, 0.2, 0.3f);
         ParsedSchemaMetadata keySchemaMetadata = new ParsedSchemaMetadata(10, 2, keySchema);
         ParsedSchemaMetadata valueSchemaMetadata = new ParsedSchemaMetadata(10, 2, valueSchema);
@@ -193,13 +193,13 @@ public void withCompression() throws IOException, InterruptedException {
         webServer.enqueue(new MockResponse()
                 .setHeader("Content-Type", "application/json; charset=utf-8")
                 .setBody("{\"offset\": 100}"));
-        Schema keySchema = MeasurementKey.getClassSchema();
+        Schema keySchema = ObservationKey.getClassSchema();
         Schema valueSchema = PhoneLight.getClassSchema();
-        AvroTopic<MeasurementKey, PhoneLight> topic = new AvroTopic<>("test",
-                keySchema, valueSchema, MeasurementKey.class, PhoneLight.class);
-        KafkaTopicSender<MeasurementKey, PhoneLight> topicSender = sender.sender(topic);
+        AvroTopic<ObservationKey, PhoneLight> topic = new AvroTopic<>("test",
+                keySchema, valueSchema, ObservationKey.class, PhoneLight.class);
+        KafkaTopicSender<ObservationKey, PhoneLight> topicSender = sender.sender(topic);
 
-        MeasurementKey key = new MeasurementKey("a", "b");
+        ObservationKey key = new ObservationKey("test", "a", "b");
         PhoneLight value = new PhoneLight(0.1, 0.2, 0.3f);
         ParsedSchemaMetadata keySchemaMetadata = new ParsedSchemaMetadata(10, 2, keySchema);
         ParsedSchemaMetadata valueSchemaMetadata = new ParsedSchemaMetadata(10, 2, valueSchema);

src/test/java/org/radarcns/topic/SensorTopicTest.java

@@ -16,13 +16,14 @@
 
 package org.radarcns.topic;
 
+import com.fasterxml.jackson.databind.annotation.JsonTypeResolver;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Type;
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.generic.GenericRecord;
 import org.junit.Test;
-import org.radarcns.key.MeasurementKey;
-import org.radarcns.phone.PhoneAcceleration;
+import org.radarcns.kafka.ObservationKey;
+import org.radarcns.passive.phone.PhoneAcceleration;
 
 import static org.junit.Assert.assertEquals;
 
@@ -34,6 +35,7 @@ public class SensorTopicTest {
     @Test
     public void workingConstructor() {
         Schema keySchema = SchemaBuilder.record("key").fields()
+                .name("projectId").type(Schema.createUnion(Schema.create(Type.NULL), Schema.create(Type.STRING))).withDefault(null)
                 .name("userId").type(Schema.create(Type.STRING)).noDefault()
                 .name("sourceId").type(Schema.create(Type.STRING)).noDefault()
                 .endRecord();
@@ -123,28 +125,28 @@ public void notARecord() {
 
     @Test
     public void parseTopic() {
-        SensorTopic<MeasurementKey, PhoneAcceleration> topic = SensorTopic.parse("test",
-                MeasurementKey.class.getName(), PhoneAcceleration.class.getName());
+        SensorTopic<ObservationKey, PhoneAcceleration> topic = SensorTopic.parse("test",
+                ObservationKey.class.getName(), PhoneAcceleration.class.getName());
 
-        SensorTopic<MeasurementKey, PhoneAcceleration> expected = new SensorTopic<>("test",
-                MeasurementKey.getClassSchema(), PhoneAcceleration.getClassSchema(),
-                MeasurementKey.class, PhoneAcceleration.class);
+        SensorTopic<ObservationKey, PhoneAcceleration> expected = new SensorTopic<>("test",
+                ObservationKey.getClassSchema(), PhoneAcceleration.getClassSchema(),
+                ObservationKey.class, PhoneAcceleration.class);
 
         assertEquals(expected, topic);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void parseUnexistingKey() {
         SensorTopic.parse("test",
-                "unexisting." + MeasurementKey.class.getName(),
+                "unexisting." + ObservationKey.class.getName(),
                 PhoneAcceleration.class.getName());
     }
 
 
     @Test(expected = IllegalArgumentException.class)
     public void parseUnexistingValue() {
         SensorTopic.parse("test",
-                MeasurementKey.class.getName(),
+                ObservationKey.class.getName(),
                 "unexisting." + PhoneAcceleration.class.getName());
     }
 }

src/test/java/org/radarcns/util/serde/KafkaAvroSerializerTest.java

@@ -33,7 +33,7 @@
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
 import org.junit.Test;
-import org.radarcns.key.MeasurementKey;
+import org.radarcns.kafka.ObservationKey;
 import org.radarcns.producer.rest.SchemaRetriever;
 import org.radarcns.producer.rest.ParsedSchemaMetadata;
 
@@ -44,7 +44,7 @@ public void serialize() throws Exception {
         testSerialization("this", Schema.create(Type.STRING));
         testSerialization(10, Schema.create(Type.INT));
 
-        MeasurementKey key = new MeasurementKey("a", "b");
+        ObservationKey key = new ObservationKey("test", "a", "b");
         testSerialization(key, key.getSchema());
 
         Schema genericSchema = Schema.createRecord(Arrays.asList(

src/test/resources/integration.yml

@@ -26,7 +26,7 @@ number_of_devices: 1
 data:
   - file: integration_test.csv
     topic: integration_test
-    key_schema: org.radarcns.key.MeasurementKey
+    key_schema: org.radarcns.kafka.ObservationKey
     value_schema: org.radarcns.aggregator.DoubleArrayAggregator
     frequency: 32
     sensor: TEMPERATURE

testing/src/main/java/org/radarcns/mock/MockFileSender.java

@@ -19,7 +19,7 @@
 import java.io.IOException;
 import org.apache.avro.specific.SpecificRecord;
 import org.radarcns.data.Record;
-import org.radarcns.key.MeasurementKey;
+import org.radarcns.kafka.ObservationKey;
 import org.radarcns.mock.data.MockCsvParser;
 import org.radarcns.producer.KafkaSender;
 import org.radarcns.producer.KafkaTopicSender;
@@ -33,7 +33,7 @@ public class MockFileSender {
     private final KafkaSender sender;
     private final MockCsvParser parser;
 
-    public MockFileSender(KafkaSender<MeasurementKey, SpecificRecord> sender,
+    public MockFileSender(KafkaSender<ObservationKey, SpecificRecord> sender,
             MockCsvParser parser) {
         this.parser = parser;
         this.sender = sender;