fixes bug delete with pattern filtering (#366)

Author: conker84

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

@@ -126,13 +126,11 @@ class StreamsTransactionEventHandler(private val router: StreamsEventRouter,
                 .toMap()
 
         // returns a Map<Boolean, List<Node>> where the K is true if the node has been deleted
-        val removedNodeProps = allOrFiltered(txd.removedNodeProperties(), nodeAll)
-                { it.entity().labelNames().any { nodeRoutingLabels.contains(it) } }
+        val removedNodeProps = txd.removedNodeProperties()
                 .map { txd.deletedNodes().contains(it.entity()) to it }
                 .groupBy({ it.first }, { it.second })
                 .toMap()
-        val removedLbls = allOrFiltered(txd.removedLabels(), nodeAll)
-                { nodeRoutingLabels.contains(it.label().name()) }
+        val removedLbls = txd.removedLabels()
                 .map { txd.deletedNodes().contains(it.node()) to it }
                 .groupBy({ it.first }, { it.second })
                 .toMap()
@@ -147,12 +145,10 @@ class StreamsTransactionEventHandler(private val router: StreamsEventRouter,
                 .map { labelEntry -> labelEntry.node().id to labelEntry.label().name() } // [ (nodeId, [label]) ]
                 .groupBy({it.first},{it.second}) // { nodeId -> [label]  }
 
-
         val removedNodeProperties = removedNodeProps.getOrDefault(false, emptyList())
         val removedLabels = removedLbls.getOrDefault(false, emptyList())
 
-        val deletedPayload = allOrFiltered(txd.deletedNodes(), nodeAll)
-                { it.labels.any { nodeRoutingLabels.contains(it.name()) } }
+        val deletedPayload = txd.deletedNodes()
                 .map {
                     val beforeNode = NodeChangeBuilder()
                             .withLabels(deletedLabels.getOrDefault(it.id, emptyList()))

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

@@ -2,7 +2,10 @@ package streams.integrations
 
 import kotlinx.coroutines.async
 import kotlinx.coroutines.runBlocking
+import org.hamcrest.Matchers
 import org.junit.Test
+import org.neo4j.function.ThrowingSupplier
+import streams.Assert
 import streams.events.EntityType
 import streams.events.NodeChange
 import streams.events.NodePayload
@@ -14,6 +17,7 @@ import streams.KafkaTestUtils
 import streams.serialization.JSONUtils
 import streams.setConfig
 import streams.start
+import java.util.concurrent.TimeUnit
 import kotlin.test.assertEquals
 
 class KafkaEventRouterSimpleTSE: KafkaEventRouterBaseTSE() {
@@ -131,4 +135,36 @@ class KafkaEventRouterSimpleTSE: KafkaEventRouterBaseTSE() {
             }
         })
     }
+
+    @Test
+    fun testDeleteNode() = runBlocking {
+        val topic = "testdeletetopic"
+        db.setConfig("streams.source.topic.nodes.$topic.from.neo4j", "Person:Neo4j{*}")
+            .setConfig("streams.source.topic.relationships.$topic.from.neo4j", "KNOWS{*}")
+            .start()
+        kafkaConsumer.subscribe(listOf(topic))
+        db.execute("CREATE (:Person:ToRemove {name:'John Doe', age:42})-[:KNOWS]->(:Person {name:'Jane Doe', age:36})")
+        Assert.assertEventually(ThrowingSupplier {
+            kafkaConsumer.poll(5000).count() > 0
+        }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+        db.execute("MATCH (p:Person {name:'John Doe', age:42}) REMOVE p:ToRemove")
+        Assert.assertEventually(ThrowingSupplier {
+            kafkaConsumer.poll(5000)
+                    .map { JSONUtils.asStreamsTransactionEvent(it.value()) }
+                    .filter { it.meta.operation == OperationType.updated }
+                    .count() > 0
+        }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+        db.execute("MATCH (p:Person) DETACH DELETE p")
+        val count = db.execute("MATCH (p:Person {name:'John Doe', age:42}) RETURN count(p) AS count") {
+            it.columnAs<Long>("count").next()
+        }
+        assertEquals(0, count)
+        Assert.assertEventually(ThrowingSupplier {
+            val count = kafkaConsumer.poll(5000)
+                .map { JSONUtils.asStreamsTransactionEvent(it.value()) }
+                .filter { it.meta.operation == OperationType.deleted }
+                .count()
+            count > 0
+        }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+    }
 }