Rebase v 3.5 to the last version of master branch (#106)

Author: conker84

common/pom.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.neo4j</groupId>
+    <artifactId>neo4j-streams-common</artifactId>
+    <name>Neo4j Streams - Common</name>
+    <description>Neo4j Streams - Commons Package</description>
+    <version>3.5.0</version>
+    <packaging>jar</packaging>
+
+    <parent>
+        <groupId>org.neo4j</groupId>
+        <artifactId>neo4j-streams-parent</artifactId>
+        <version>3.5.0</version>
+    </parent>
+
+</project>

common/src/main/kotlin/streams/extensions/CommonExtensions.kt

@@ -0,0 +1,13 @@
+package streams.extensions
+
+import org.neo4j.graphdb.Node
+
+fun Map<String,String>.getInt(name:String, defaultValue: Int) = this.get(name)?.toInt() ?: defaultValue
+
+fun Node.labelNames() : List<String> {
+    return this.labels.map { it.name() }
+}
+
+fun String.toPointCase(): String {
+    return this.split("(?<=[a-z])(?=[A-Z])".toRegex()).joinToString(separator = ".").toLowerCase()
+}

producer/src/main/kotlin/streams/serialization/JacksonUtil.kt renamed to common/src/main/kotlin/streams/utils/JSONUtils.kt

@@ -52,7 +52,7 @@ class TemporalAccessorSerializer : JsonSerializer<TemporalAccessor>() {
 }
 
 
-object JacksonUtil {
+object JSONUtils {
 
     private val OBJECT_MAPPER: ObjectMapper = ObjectMapper()
 
@@ -64,8 +64,24 @@ object JacksonUtil {
         OBJECT_MAPPER.disable(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS)
     }
 
-    fun getMapper(): ObjectMapper {
+    fun getObjectMapper(): ObjectMapper {
         return OBJECT_MAPPER
     }
 
+    fun asMap(any: Any): Map<String, Any?> {
+        return OBJECT_MAPPER.convertValue(any, Map::class.java)
+                .mapKeys { it.key.toString() }
+    }
+
+    fun writeValueAsString(any: Any): String {
+        return OBJECT_MAPPER.writeValueAsString(any)
+    }
+
+    fun writeValueAsBytes(any: Any): ByteArray {
+        return OBJECT_MAPPER.writeValueAsBytes(any)
+    }
+
+    fun <T> readValue(value: ByteArray, clazz: Class<T>): T {
+        return OBJECT_MAPPER.readValue(value, clazz)
+    }
 }

consumer/src/main/kotlin/streams/utils/Neo4jUtils.kt renamed to common/src/main/kotlin/streams/utils/Neo4jUtils.kt

@@ -1,8 +1,8 @@
 package streams.utils
 
 import org.neo4j.graphdb.QueryExecutionException
-import org.neo4j.kernel.impl.logging.LogService
 import org.neo4j.kernel.internal.GraphDatabaseAPI
+import org.neo4j.logging.internal.LogService
 import java.lang.reflect.InvocationTargetException
 
 object Neo4jUtils {

consumer/src/main/kotlin/streams/utils/StreamsUtils.kt renamed to common/src/main/kotlin/streams/utils/StreamsUtils.kt

@@ -2,11 +2,20 @@ package streams.utils
 
 object StreamsUtils {
 
+    const val UNWIND: String = "UNWIND {events} AS event"
+
+    const val STREAMS_CONFIG_PREFIX = "streams."
+
+    const val STREAMS_SINK_TOPIC_PREFIX = "sink.topic.cypher."
+
     fun <T> ignoreExceptions(action: () -> T, vararg toIgnore: Class<out Throwable>): T? {
         return try {
             action()
         } catch (e: Throwable) {
-            when (e::class.java) {
+            if (toIgnore.isEmpty()) {
+                return null
+            }
+            return when (e::class.java) {
                 in toIgnore -> null
                 else -> throw e
             }

producer/src/test/kotlin/streams/serialization/JacksonUtilTest.kt renamed to common/src/test/kotlin/streams/utils/JSONUtilsTest.kt

@@ -9,7 +9,7 @@ import org.neo4j.values.storable.Values.pointValue
 import java.time.ZoneOffset.UTC
 import kotlin.test.assertEquals
 
-class JacksonUtilTest {
+class JSONUtilsTest {
 
     @Test
     fun shouldSerializeGeometryAndTemporalDataTypes() {
@@ -27,7 +27,7 @@ class JacksonUtilTest {
                 "dateTime" to datetime(date( 2017, 12, 17), time( 17, 14, 35, 123456789, UTC)))
 
         // When
-        val jsonString = JacksonUtil.getMapper().writeValueAsString(map)
+        val jsonString = JSONUtils.writeValueAsString(map)
 
         // Then
         assertEquals(expected, jsonString)

consumer/src/test/kotlin/streams/utils/Neo4jUtilsTest.kt renamed to common/src/test/kotlin/streams/utils/Neo4jUtilsTest.kt

@@ -1,7 +1,8 @@
 package streams.utils
 
-import org.junit.*
-import org.neo4j.kernel.impl.logging.LogService
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.test.TestGraphDatabaseFactory
 import kotlin.test.assertTrue

consumer/src/test/kotlin/streams/utils/StreamsUtilsTest.kt renamed to common/src/test/kotlin/streams/utils/StreamsUtilsTest.kt

@@ -6,16 +6,23 @@
 
     <groupId>org.neo4j</groupId>
     <artifactId>neo4j-streams-consumer</artifactId>
-    <version>3.4.0</version>
+    <version>3.5.0</version>
     <name>Neo4j Streams - Consumer</name>
     <description>Neo4j Streams - Kafka Consumer</description>
     <packaging>jar</packaging>
 
     <parent>
         <groupId>org.neo4j</groupId>
         <artifactId>neo4j-streams-parent</artifactId>
-        <version>3.4.0</version>
+        <version>3.5.0</version>
     </parent>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.neo4j</groupId>
+            <artifactId>neo4j-streams-common</artifactId>
+            <version>3.5.0</version>
+        </dependency>
+    </dependencies>
 
 </project>

consumer/pom.xml

@@ -1,33 +1,51 @@
 package streams
 
-import org.neo4j.kernel.AvailabilityGuard
 import org.neo4j.kernel.configuration.Config
-import org.neo4j.kernel.internal.GraphDatabaseAPI
+import org.neo4j.logging.Log
 
 abstract class StreamsEventSink(private val config: Config,
-                                private val db: GraphDatabaseAPI): AvailabilityGuard.AvailabilityListener {
+                                private val queryExecution: StreamsEventSinkQueryExecution,
+                                private val streamsTopicService: StreamsTopicService,
+                                private val log: Log) {
 
-    abstract var streamsTopicService: StreamsTopicService?
+    abstract fun stop()
+
+    abstract fun start()
+
+    abstract fun getEventConsumerFactory(): StreamsEventConsumerFactory
+
+    abstract fun getEventSinkConfigMapper(): StreamsEventSinkConfigMapper
+
+}
+
+abstract class StreamsEventConsumer<T>(private val consumer: T, config: StreamsSinkConfiguration, private val log: Log) {
 
     abstract fun stop()
 
+    abstract fun withTopics(topics: Set<String>): StreamsEventConsumer<T>
+
     abstract fun start()
 
-    override fun unavailable() {
-        stop()
-    }
+    abstract fun read(): Map<String, List<Map<String, Any?>>>?
 
-    override fun available() {
-        start()
-    }
+}
 
+abstract class StreamsEventConsumerFactory {
+    abstract fun createStreamsEventConsumer(config: Map<String, String>, log: Log): StreamsEventConsumer<*>
 }
 
 object StreamsEventSinkFactory {
-    fun getStreamsEventSink(config: Config, db: GraphDatabaseAPI): StreamsEventSink {
+    fun getStreamsEventSink(config: Config, streamsQueryExecution: StreamsEventSinkQueryExecution,
+                            streamsTopicService: StreamsTopicService, log: Log): StreamsEventSink {
         return Class.forName(config.raw.getOrDefault("streams.sink", "streams.kafka.KafkaEventSink"))
                 .getConstructor(Config::class.java,
-                        GraphDatabaseAPI::class.java)
-                .newInstance(config, db) as StreamsEventSink
+                        StreamsEventSinkQueryExecution::class.java,
+                        StreamsTopicService::class.java,
+                        Log::class.java)
+                .newInstance(config, streamsQueryExecution, streamsTopicService, log) as StreamsEventSink
     }
+}
+
+abstract class StreamsEventSinkConfigMapper(private val baseConfiguration: Map<String, String>, private val mapping: Map<String, String>) {
+    abstract fun convert(config: Map<String, String>): Map<String, String>
 }