Fixes #466: Problem of streams.publish procedure (#469)

Author: vga91

producer/src/main/kotlin/streams/Extensions.kt

@@ -15,7 +15,7 @@ fun Node.toMap(): Map<String, Any?> {
 }
 
 fun Relationship.toMap(): Map<String, Any?> {
-    return mapOf("id" to id.toString(), "properties" to allProperties, "label" to type,
+    return mapOf("id" to id.toString(), "properties" to allProperties, "label" to type.name(),
             "start" to startNode.toMap(),
             "end" to endNode.toMap(),
             "type" to EntityType.relationship)

producer/src/test/kotlin/streams/integrations/KafkaEventRouterProcedureTSE.kt

@@ -4,6 +4,7 @@ import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.NewTopic
 import org.junit.Test
 import org.neo4j.graphdb.QueryExecutionException
+import org.neo4j.graphdb.Result
 import streams.events.StreamsEvent
 import streams.extensions.execute
 import streams.utils.JSONUtils
@@ -121,14 +122,7 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
                 "CALL streams.publish.sync('neo4j', n) \n" +
                 "YIELD value \n" +
                 "RETURN value") {
-            assertTrue { it.hasNext() }
-            val resultMap = (it.next())["value"] as Map<String, Any>
-            assertNotNull(resultMap["offset"])
-            assertNotNull(resultMap["partition"])
-            assertNotNull(resultMap["keySize"])
-            assertNotNull(resultMap["valueSize"])
-            assertNotNull(resultMap["timestamp"])
-            assertFalse { it.hasNext() }
+            assertSyncResult(it)
         }
 
         val records = kafkaConsumer.poll(5000)
@@ -143,14 +137,7 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
         setUpProcedureTests()
         val message = "Hello World"
         db.execute("CALL streams.publish.sync('neo4j', '$message')") {
-            assertTrue { it.hasNext() }
-            val resultMap = (it.next())["value"] as Map<String, Any>
-            assertNotNull(resultMap["offset"])
-            assertNotNull(resultMap["partition"])
-            assertNotNull(resultMap["keySize"])
-            assertNotNull(resultMap["valueSize"])
-            assertNotNull(resultMap["timestamp"])
-            assertFalse { it.hasNext() }
+            assertSyncResult(it)
         }
 
         val records = kafkaConsumer.poll(5000)
@@ -160,6 +147,38 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
         }}
     }
 
+
+    @Test
+    fun testProcedureWithRelationship() {
+        setUpProcedureTests()
+        db.execute("CREATE (:Foo {one: 'two'})-[:KNOWS {alpha: 'beta'}]->(:Bar {three: 'four'})")
+        val recordsCreation = kafkaConsumer.poll(5000)
+        assertEquals(3, recordsCreation.count())
+
+        db.execute("""
+            MATCH (:Foo)-[r:KNOWS]->(:Bar)
+            |CALL streams.publish.sync('neo4j', r)
+            |YIELD value RETURN value""".trimMargin()) {
+            assertSyncResult(it)
+        }
+        val records = kafkaConsumer.poll(5000)
+        assertEquals(1, records.count())
+        
+        val payload = JSONUtils.readValue<StreamsEvent>(records.first().value()).payload as Map<String, Any>
+        assertTrue(payload["id"] is String)
+        assertEquals(mapOf("alpha" to "beta"), payload["properties"])
+        assertEquals("KNOWS", payload["label"])
+        assertEquals("relationship", payload["type"])
+        val start = payload["start"] as Map<String, Any>
+        assertEquals(listOf("Foo"), start["labels"])
+        assertEquals(mapOf("one" to "two"), start["properties"])
+        assertEquals("node", start["type"])
+        val end = payload["end"] as Map<String, Any>
+        assertEquals(listOf("Bar"), end["labels"])
+        assertEquals(mapOf("three" to "four"), end["properties"])
+        assertEquals("node", end["type"])
+    }
+
     @Test
     fun testProcedureSyncWithKeyNull() {
         setUpProcedureTests()
@@ -169,15 +188,8 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
         assertEquals(1, recordsCreation.count())
 
         val message = "Hello World"
-            db.execute("MATCH (n:Foo {id: 1}) CALL streams.publish.sync('neo4j', '$message', {key: n.foo}) YIELD value RETURN value") {
-            assertTrue { it.hasNext() }
-            val resultMap = (it.next())["value"] as Map<String, Any>
-            assertNotNull(resultMap["offset"])
-            assertNotNull(resultMap["partition"])
-            assertNotNull(resultMap["keySize"])
-            assertNotNull(resultMap["valueSize"])
-            assertNotNull(resultMap["timestamp"])
-            assertFalse { it.hasNext() }
+        db.execute("MATCH (n:Foo {id: 1}) CALL streams.publish.sync('neo4j', '$message', {key: n.foo}) YIELD value RETURN value") { 
+            assertSyncResult(it)
         }
 
         val records = kafkaConsumer.poll(5000)
@@ -204,14 +216,7 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
             val keyRecord = "test"
             val partitionRecord = 1
             db.execute("CALL streams.publish.sync('$topic', '$message', {key: '$keyRecord', partition: $partitionRecord })") {
-                assertTrue { it.hasNext() }
-                val resultMap = (it.next())["value"] as Map<String, Any>
-                assertNotNull(resultMap["offset"])
-                assertEquals(partitionRecord, resultMap["partition"])
-                assertNotNull(resultMap["keySize"])
-                assertNotNull(resultMap["valueSize"])
-                assertNotNull(resultMap["timestamp"])
-                assertFalse { it.hasNext() }
+                assertSyncResult(it)
             }
 
             val records = kafkaConsumer.poll(5000)
@@ -280,4 +285,15 @@ class KafkaEventRouterProcedureTSE : KafkaEventRouterBaseTSE() {
         KafkaEventRouterSuiteIT.registerPublishProcedure(db)
         kafkaConsumer.subscribe(listOf("neo4j"))
     }
+
+    private fun assertSyncResult(it: Result) {
+        assertTrue { it.hasNext() }
+        val resultMap = (it.next())["value"] as Map<String, Any>
+        assertNotNull(resultMap["offset"])
+        assertNotNull(resultMap["partition"])
+        assertNotNull(resultMap["keySize"])
+        assertNotNull(resultMap["valueSize"])
+        assertNotNull(resultMap["timestamp"])
+        assertFalse { it.hasNext() }
+    }
 }