fixes #104 (#105)

Author: conker84

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

@@ -1,12 +1,20 @@
 package streams
 
+import org.apache.commons.lang3.StringUtils
 import streams.events.EntityType
 
 
-private fun <T> filterMap(config: Map<String, String>, routingPrefix: String): List<T> {
+private inline fun <reified T> filterMap(config: Map<String, String>, routingPrefix: String): List<T> {
+    val entityType = when (T::class) {
+        NodeRoutingConfiguration::class -> EntityType.node
+        RelationshipRoutingConfiguration::class -> EntityType.relationship
+        else -> throw IllegalArgumentException("The class must be an instance of RoutingConfiguration")
+    }
     return config
             .filterKeys { it.startsWith(routingPrefix) }
-            .flatMap { RoutingConfigurationFactory.getRoutingConfiguration(it.key.replace(routingPrefix, "") , it.value, EntityType.node) as List<T> }
+            .flatMap { RoutingConfigurationFactory
+                    .getRoutingConfiguration(it.key.replace(routingPrefix, StringUtils.EMPTY) , it.value, entityType) as List<T>
+            }
 }
 
 private object StreamsRoutingConfigurationConstants {

producer/src/test/kotlin/streams/RoutingConfigurationTest.kt

@@ -8,9 +8,24 @@ import kotlin.test.assertTrue
 
 class RoutingConfigurationTest {
 
-    @Test(expected = IllegalArgumentException::class)
+    @Test
     fun badPatternShouldThrowIllegalArgumentException() {
-        RoutingConfigurationFactory.getRoutingConfiguration("topic1", "Label(1,2)", EntityType.node)
+        val topic = "topic1"
+        assertIllegalArgumentException(topic, "Label(1,2)", EntityType.node)
+        assertIllegalArgumentException(topic, "Label{}", EntityType.node)
+        assertIllegalArgumentException(topic, "KNOWS{}", EntityType.relationship)
+    }
+
+    private fun assertIllegalArgumentException(topic: String, pattern: String, entityType: EntityType) {
+        var hasException = false
+        try {
+            RoutingConfigurationFactory.getRoutingConfiguration(topic, pattern, entityType)
+        } catch (e: Exception) {
+            assertTrue { e is IllegalArgumentException }
+            assertEquals("The pattern $pattern for topic $topic is invalid", e.message)
+            hasException = true
+        }
+        assertTrue { hasException }
     }
 
     @Test

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

@@ -4,10 +4,8 @@ import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.apache.kafka.common.serialization.StringDeserializer
-import org.junit.After
-import org.junit.Before
-import org.junit.ClassRule
-import org.junit.Test
+import org.junit.*
+import org.junit.rules.TestName
 import org.neo4j.kernel.impl.proc.Procedures
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.test.TestGraphDatabaseFactory
@@ -39,11 +37,26 @@ class KafkaEventRouterIT {
 
     lateinit var db: GraphDatabaseAPI
 
+    private val WITH_REL_ROUTING_METHOD_SUFFIX = "WithRelRouting"
+    private val WITH_NODE_ROUTING_METHOD_SUFFIX = "WithNodeRouting"
+
+    @Rule
+    @JvmField
+    var testName = TestName()
+
     @Before
     fun setUp() {
-        db = TestGraphDatabaseFactory().newImpermanentDatabaseBuilder()
+        var graphDatabaseBuilder = TestGraphDatabaseFactory()
+                .newImpermanentDatabaseBuilder()
                 .setConfig("kafka.bootstrap.servers", kafka.bootstrapServers)
-                .newGraphDatabase() as GraphDatabaseAPI
+        if (testName.methodName.endsWith(WITH_REL_ROUTING_METHOD_SUFFIX)) {
+            graphDatabaseBuilder.setConfig("streams.source.topic.relationships.knows", "KNOWS{*}")
+        }
+        if (testName.methodName.endsWith(WITH_NODE_ROUTING_METHOD_SUFFIX)) {
+            graphDatabaseBuilder.setConfig("streams.source.topic.nodes.person", "Person{*}")
+        }
+        db = graphDatabaseBuilder.newGraphDatabase() as GraphDatabaseAPI
+
         db.dependencyResolver.resolveDependency(Procedures::class.java)
                 .registerProcedure(StreamsProcedures::class.java, true)
 
@@ -76,6 +89,46 @@ class KafkaEventRouterIT {
         consumer.close()
     }
 
+    @Test
+    fun testCreateRelationshipWithRelRouting() {
+        val config = KafkaConfiguration(bootstrapServers = kafka.bootstrapServers)
+        val consumer = createConsumer(config)
+        consumer.subscribe(listOf("knows"))
+        db.execute("CREATE (:Person {name:'Andrea'})-[:KNOWS{since: 2014}]->(:Person {name:'Michael'})").close()
+        val records = consumer.poll(5000)
+        assertEquals(1, records.count())
+        assertEquals(true, records.all {
+            JSONUtils.readValue(it.value(), Map::class.java).let {
+                var payload = it["payload"] as Map<String, Any?>
+                val after = payload["after"] as Map<String, Any?>
+                val properties = after["properties"] as Map<String, Any?>
+                payload["type"] == "relationship" && payload["label"] == "KNOWS" && properties["since"] == 2014
+            }
+        })
+        consumer.close()
+    }
+
+    @Test
+    fun testCreateNodeWithNodeRouting() {
+        val config = KafkaConfiguration(bootstrapServers = kafka.bootstrapServers)
+        val consumer = createConsumer(config)
+        consumer.subscribe(listOf("person"))
+        db.execute("CREATE (:Person {name:'Andrea'})").close()
+        val records = consumer.poll(5000)
+        assertEquals(1, records.count())
+        assertEquals(true, records.all {
+            JSONUtils.readValue(it.value(), Map::class.java).let {
+                var payload = it["payload"] as Map<String, Any?>
+                val after = payload["after"] as Map<String, Any?>
+                val labels = after["labels"] as List<String>
+                val propertiesAfter = after["properties"] as Map<String, Any?>
+                val meta = it["meta"] as Map<String, Any?>
+                labels == listOf("Person") && propertiesAfter == mapOf("name" to "Andrea") && meta["operation"] == "created"
+            }
+        })
+        consumer.close()
+    }
+
     private fun createConsumer(config: KafkaConfiguration): KafkaConsumer<String, ByteArray> {
         val props = config.asProperties()
         props["group.id"] = "neo4j"