fixes #14: Add schema reference into Data Events (#151)

Author: conker84

common/pom.xml

@@ -17,4 +17,13 @@
         <version>3.5.0</version>
     </parent>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.neo4j.driver</groupId>
+            <artifactId>neo4j-java-driver</artifactId>
+            <version>${neo4j.java.driver.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
 </project>

producer/src/main/kotlin/streams/events/StreamsEvent.kt renamed to common/src/main/kotlin/streams/events/StreamsEvent.kt

@@ -16,7 +16,8 @@ data class Meta(val timestamp: Long,
 enum class EntityType { node, relationship }
 
 data class RelationshipNodeChange(val id: String,
-                                  val labels: List<String>?)
+                                  val labels: List<String>?,
+                                  val ids: Map<String, Any>)
 
 abstract class RecordChange{ abstract val properties: Map<String, Any>? }
 data class NodeChange(override val properties: Map<String, Any>?,
@@ -31,24 +32,26 @@ abstract class Payload {
     abstract val after: RecordChange?
 }
 data class NodePayload(override val id: String,
-                       override val before: RecordChange?,
-                       override val after: RecordChange?,
+                       override val before: NodeChange?,
+                       override val after: NodeChange?,
                        override val type: EntityType = EntityType.node): Payload()
 
 data class RelationshipPayload(override val id: String,
                                val start: RelationshipNodeChange,
                                val end: RelationshipNodeChange,
-                               override val before: RecordChange?,
-                               override val after: RecordChange?,
+                               override val before: RelationshipChange?,
+                               override val after: RelationshipChange?,
                                val label: String,
                                override val type: EntityType = EntityType.relationship): Payload()
 
+enum class StreamsConstraintType { UNIQUE, NODE_PROPERTY_EXISTS, RELATIONSHIP_PROPERTY_EXISTS }
+
 data class Constraint(val label: String?,
-                      val properties: List<String>,
-                      val type: ConstraintType)
+                      val properties: Set<String>,
+                      val type: StreamsConstraintType)
 
-data class Schema(val properties: List<String> = emptyList(),
-                  val constraints: List<Constraint>? = null)
+data class Schema(val properties: Map<String, String> = emptyMap(),
+                  val constraints: List<Constraint> = emptyList())
 
 open class StreamsEvent(open val payload: Any)
 data class StreamsTransactionEvent(val meta: Meta, override val payload: Payload, val schema: Schema): StreamsEvent(payload)

common/src/main/kotlin/streams/utils/JSONUtils.kt

@@ -1,15 +1,20 @@
 package streams.serialization
 
-import org.codehaus.jackson.JsonGenerator
-import org.codehaus.jackson.JsonProcessingException
-import org.codehaus.jackson.Version
-import org.codehaus.jackson.map.JsonSerializer
-import org.codehaus.jackson.map.ObjectMapper
-import org.codehaus.jackson.map.SerializationConfig
-import org.codehaus.jackson.map.SerializerProvider
-import org.codehaus.jackson.map.module.SimpleModule
+import com.fasterxml.jackson.core.JsonGenerator
+import com.fasterxml.jackson.core.JsonProcessingException
+import com.fasterxml.jackson.databind.JsonSerializer
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.SerializationFeature
+import com.fasterxml.jackson.databind.SerializerProvider
+import com.fasterxml.jackson.databind.module.SimpleModule
+import com.fasterxml.jackson.module.kotlin.convertValue
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.fasterxml.jackson.module.kotlin.readValue
+import org.neo4j.driver.internal.value.PointValue
 import org.neo4j.graphdb.spatial.Point
 import org.neo4j.values.storable.CoordinateReferenceSystem
+import streams.events.*
+import streams.utils.StreamsUtils
 import java.io.IOException
 import java.time.temporal.TemporalAccessor
 
@@ -29,6 +34,18 @@ fun Point.toStreamsPoint(): StreamsPoint {
     }
 }
 
+fun PointValue.toStreamsPoint(): StreamsPoint {
+    val point = this.asPoint()
+    val crsType = point.srid()
+    return when (crsType) {
+        CoordinateReferenceSystem.Cartesian.code -> StreamsPointCartesian(CoordinateReferenceSystem.Cartesian.name, point.x(), point.y())
+        CoordinateReferenceSystem.Cartesian_3D.code -> StreamsPointCartesian(CoordinateReferenceSystem.Cartesian_3D.name, point.x(), point.y(), point.z())
+        CoordinateReferenceSystem.WGS84.code -> StreamsPointWgs(CoordinateReferenceSystem.WGS84.name, point.x(), point.y())
+        CoordinateReferenceSystem.WGS84_3D.code -> StreamsPointWgs(CoordinateReferenceSystem.WGS84_3D.name, point.x(), point.y(), point.z())
+        else -> throw IllegalArgumentException("Point type $crsType not supported")
+    }
+}
+
 class PointSerializer : JsonSerializer<Point>() {
     @Throws(IOException::class, JsonProcessingException::class)
     override fun serialize(value: Point?, jgen: JsonGenerator,
@@ -40,6 +57,17 @@ class PointSerializer : JsonSerializer<Point>() {
     }
 }
 
+class PointValueSerializer : JsonSerializer<PointValue>() {
+    @Throws(IOException::class, JsonProcessingException::class)
+    override fun serialize(value: PointValue?, jgen: JsonGenerator,
+                           provider: SerializerProvider) {
+        if (value == null) {
+            return
+        }
+        jgen.writeObject(value.toStreamsPoint())
+    }
+}
+
 class TemporalAccessorSerializer : JsonSerializer<TemporalAccessor>() {
     @Throws(IOException::class, JsonProcessingException::class)
     override fun serialize(value: TemporalAccessor?, jgen: JsonGenerator,
@@ -54,14 +82,15 @@ class TemporalAccessorSerializer : JsonSerializer<TemporalAccessor>() {
 
 object JSONUtils {
 
-    private val OBJECT_MAPPER: ObjectMapper = ObjectMapper()
+    private val OBJECT_MAPPER: ObjectMapper = jacksonObjectMapper()
 
     init {
-        val module = SimpleModule("Neo4jKafkaSerializer", Version(1, 0, 0, ""))
-        module.addSerializer(Point::class.java, PointSerializer())
+        val module = SimpleModule("Neo4jKafkaSerializer")
+        StreamsUtils.ignoreExceptions({ module.addSerializer(Point::class.java, PointSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
+        StreamsUtils.ignoreExceptions({ module.addSerializer(PointValue::class.java, PointValueSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
         module.addSerializer(TemporalAccessor::class.java, TemporalAccessorSerializer())
         OBJECT_MAPPER.registerModule(module)
-        OBJECT_MAPPER.disable(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS)
+        OBJECT_MAPPER.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
     }
 
     fun getObjectMapper(): ObjectMapper {
@@ -81,7 +110,50 @@ object JSONUtils {
         return OBJECT_MAPPER.writeValueAsBytes(any)
     }
 
-    fun <T> readValue(value: ByteArray, clazz: Class<T>): T {
-        return OBJECT_MAPPER.readValue(value, clazz)
+    inline fun <reified T> readValue(value: ByteArray): T {
+        return getObjectMapper().readValue(value, T::class.java)
+    }
+
+    inline fun <reified T> readValue(value: Any): T {
+        val strValue = when (value) {
+            is String -> value
+            else -> getObjectMapper().writeValueAsString(value)
+        }
+        return getObjectMapper().readValue(strValue)
+    }
+
+    inline fun <reified T> convertValue(value: Any): T {
+        return getObjectMapper().convertValue(value)
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    fun asStreamsTransactionEvent(obj: Any): StreamsTransactionEvent {
+        val value = when (obj) {
+            is Map<*, *> -> obj as Map<String, Map<String, Any>>
+            is String -> readValue(obj)
+            is ByteArray -> readValue(obj)
+            else -> convertValue(obj)
+        }
+        val meta = convertValue<Meta>(value.getValue("meta"))
+
+        val schema = convertValue<Schema>(value.getValue("schema"))
+
+        val payloadMap = value.getValue("payload")
+        val type = payloadMap.getValue("type").toString()
+        val id = payloadMap.getValue("id").toString()
+
+        val payload = if (type == "node") {
+            val before = if (payloadMap["before"] != null) convertValue<NodeChange>(payloadMap["before"]!!) else null
+            val after = if (payloadMap["after"] != null) convertValue<NodeChange>(payloadMap["after"]!!) else null
+            NodePayload(id, before, after)
+        } else {
+            val label= payloadMap.getValue("label").toString()
+            val start = convertValue<RelationshipNodeChange>(payloadMap.getValue("start"))
+            val end = convertValue<RelationshipNodeChange>(payloadMap.getValue("end"))
+            val before = if (payloadMap["before"] != null) convertValue<RelationshipChange>(payloadMap["before"]!!) else null
+            val after = if (payloadMap["after"] != null) convertValue<RelationshipChange>(payloadMap["after"]!!) else null
+            RelationshipPayload(id, start, end, before, after, label)
+        }
+        return StreamsTransactionEvent(meta, payload, schema)
     }
 }

consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt

@@ -120,7 +120,7 @@ class KafkaEventConsumer(private val consumer: KafkaConsumer<String, ByteArray>,
         if (records != null && !records.isEmpty) {
             return records
                     .map {
-                        it.topic()!! to JSONUtils.readValue(it.value(), Any::class.java)
+                        it.topic()!! to JSONUtils.readValue<Any>(it.value())
                     }
                     .groupBy({ it.first }, { it.second })
         }

doc/asciidoc/producer/configuration.adoc

@@ -19,6 +19,7 @@ kafka.transactional.id=
 streams.source.topic.nodes.<TOPIC_NAME>=<PATTERN>
 streams.source.topic.relationships.<TOPIC_NAME>=<PATTERN>
 streams.source.enable=<true/false, default=true>
+streams.source.schema.polling.interval=<MILLIS, the polling interval for getting the schema information>
 ----
 
 Note: To use the Kafka transactions please set `kafka.transactional.id` and `kafka.acks` properly

doc/asciidoc/producer/data/node.created.json

@@ -16,10 +16,22 @@
     "after": {
       "labels": ["Person"],
       "properties": {
-        "last_name": "Kretchmar",
         "email": "[email protected]",
+        "last_name": "Kretchmar",
         "first_name": "Anne Marie"
       }
     }
+  },
+  "schema": {
+    "properties": {
+      "last_name": "String",
+      "email": "String",
+      "first_name": "String"
+    },
+    "constraints": [{
+      "label": "Person",
+      "properties": ["first_name", "last_name"],
+      "type": "UNIQUE"
+    }]
   }
 }

doc/asciidoc/producer/data/node.deleted.json

@@ -19,8 +19,21 @@
         "last_name": "Kretchmar",
         "email": "[email protected]",
         "first_name": "Anne Marie",
-        "geo":[0.123, 46.2222, 32.11111]
+        "geo": { "crs": "wgs-84-3d", "latitude": 46.2222, "longitude": 32.11111, "height": 0.123 }
       }
     }
+  },
+  "schema": {
+    "properties": {
+      "last_name": "String",
+      "email": "String",
+      "first_name": "String",
+      "geo": "point"
+    },
+    "constraints": [{
+      "label": "Person",
+      "properties": ["first_name", "last_name"],
+      "type": "UNIQUE"
+    }]
   }
 }

doc/asciidoc/producer/data/node.updated.json

@@ -27,8 +27,20 @@
         "last_name": "Kretchmar",
         "email": "[email protected]",
         "first_name": "Anne Marie",
-        "geo":[0.123, 46.2222, 32.11111]
+        "geo": { "crs": "wgs-84-3d", "latitude": 46.2222, "longitude": 32.11111, "height": 0.123 }
       }
     }
+  },
+  "schema": {
+    "properties": {
+      "last_name": "String",
+      "email": "String",
+      "first_name": "String"
+    },
+    "constraints": [{
+      "label": "Person",
+      "properties": ["first_name", "last_name"],
+      "type": "UNIQUE"
+    }]
   }
 }

doc/asciidoc/producer/data/relationship.created.json

@@ -16,16 +16,34 @@
     "label": "KNOWS",
     "start": {
       "labels": ["Person"],
-      "id": "123"
+      "id": "123",
+      "ids": {
+        "last_name": "Andrea",
+        "first_name": "Santurbano"
+      }
     },
     "end": {
       "labels": ["Person"],
-      "id": "456"
+      "id": "456",
+      "ids": {
+        "last_name": "Michael",
+        "first_name": "Hunger"
+      }
     },
     "after": {
       "properties": {
         "since": "2018-04-05T12:34:00[Europe/Berlin]"
       }
     }
+  },
+  "schema": {
+    "properties": {
+      "since": "ZonedDateTime"
+    },
+    "constraints": [{
+      "label": "KNOWS",
+      "properties": ["since"],
+      "type": "RELATIONSHIP_PROPERTY_EXISTS"
+    }]
   }
 }

doc/asciidoc/producer/data/relationship.deleted.json

@@ -16,17 +16,36 @@
     "label": "KNOWS",
     "start": {
       "labels": ["Person"],
-      "id": "123"
+      "id": "123",
+      "ids": {
+        "last_name": "Andrea",
+        "first_name": "Santurbano"
+      }
     },
     "end": {
       "labels": ["Person"],
-      "id": "456"
+      "id": "456",
+      "ids": {
+        "last_name": "Michael",
+        "first_name": "Hunger"
+      }
     },
     "before": {
       "properties": {
         "since": "2018-04-05T12:34:00[Europe/Berlin]",
         "to": "2019-04-05T23:00:00[Europe/Berlin]"
       }
     }
+  },
+  "schema": {
+    "properties": {
+      "since": "ZonedDateTime",
+      "to": "ZonedDateTime"
+    },
+    "constraints": [{
+      "label": "KNOWS",
+      "properties": ["since"],
+      "type": "RELATIONSHIP_PROPERTY_EXISTS"
+    }]
   }
 }