fixes #84: Add a stream.publish(topic, payload) procedure (#85)

Author: conker84

consumer/src/main/kotlin/streams/StreamsEventSinkQueryExecution.kt

@@ -13,7 +13,6 @@ class StreamsEventSinkQueryExecution(private val streamsTopicService: StreamsTop
             return
         }
         if(log.isDebugEnabled){
-
             log.debug("Processing ${params.size} events from Kafka")
         }
         db.execute("$UNWIND $cypherQuery", mapOf("events" to params)).close()

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

@@ -27,7 +27,7 @@ class KafkaEventSink: StreamsEventSink {
 
     private lateinit var job: Job
     private lateinit var queryExecution: StreamsEventSinkQueryExecution
-    private lateinit var kafkaConsumer: KafkaConsumer<Long, ByteArray>
+    private lateinit var kafkaConsumer: KafkaConsumer<String, ByteArray>
 
     override var streamsTopicService: StreamsTopicService? = null
 
@@ -84,7 +84,7 @@ class KafkaEventSink: StreamsEventSink {
         }
     }
 
-    private fun consume(records: ConsumerRecords<Long, ByteArray>) {
+    private fun consume(records: ConsumerRecords<String, ByteArray>) {
         streamsTopicService!!.getTopics().forEach {
             if (log.isDebugEnabled) {
                 log.debug("Reading data from topic $it")

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

@@ -3,6 +3,7 @@ package streams.kafka
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.apache.kafka.common.serialization.LongDeserializer
+import org.apache.kafka.common.serialization.StringDeserializer
 import org.codehaus.jackson.map.ObjectMapper
 import org.neo4j.kernel.configuration.Config
 import streams.StreamsSinkConfiguration
@@ -67,7 +68,7 @@ data class KafkaSinkConfiguration(val zookeeperConnect: String = "localhost:2181
 
     private fun addDeserializers() : Properties {
         val props = Properties()
-        props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = LongDeserializer::class.java
+        props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
         props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ByteArrayDeserializer::class.java
         return props
     }

consumer/src/test/kotlin/integrations/KafkaEventSinkIT.kt

@@ -10,7 +10,7 @@ import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.common.serialization.ByteArraySerializer
-import org.apache.kafka.common.serialization.LongSerializer
+import org.apache.kafka.common.serialization.StringSerializer
 import org.codehaus.jackson.map.ObjectMapper
 import org.junit.*
 import org.junit.rules.TestName
@@ -59,7 +59,7 @@ class KafkaEventSinkIT {
 
     private val kafkaProperties = Properties()
 
-    private lateinit var kafkaProducer: KafkaProducer<Long, ByteArray>
+    private lateinit var kafkaProducer: KafkaProducer<String, ByteArray>
 
     // Test data
     private val dataProperties = mapOf("prop1" to "foo", "bar" to 1)
@@ -77,7 +77,7 @@ class KafkaEventSinkIT {
 
         kafkaProperties[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = kafka.bootstrapServers
         kafkaProperties["group.id"] = "neo4j"
-        kafkaProperties[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = LongSerializer::class.java
+        kafkaProperties[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
         kafkaProperties[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java
 
         AdminClient.create(kafkaProperties).use { client -> client.createTopics(
@@ -94,7 +94,8 @@ class KafkaEventSinkIT {
     fun shouldWriteDataFromSink() = runBlocking {
         val job = GlobalScope.launch {
             var partition = ThreadLocalRandom.current().nextInt(1)
-            var producerRecord = ProducerRecord(topics[0], partition, System.currentTimeMillis(), 1L,
+            var producerRecord = ProducerRecord(topics[0], partition, System.currentTimeMillis(),
+                    UUID.randomUUID().toString(),
                     objectMapper.writeValueAsBytes(data))
             kafkaProducer.send(producerRecord).get()
             delay(5000)
@@ -126,7 +127,8 @@ class KafkaEventSinkIT {
     fun shouldNotWriteDataFromSinkWithNoTopicLoaded() = runBlocking {
         val job = GlobalScope.launch {
             var partition = ThreadLocalRandom.current().nextInt(1)
-            var producerRecord = ProducerRecord(topics[0], partition, System.currentTimeMillis(), 1L,
+            var producerRecord = ProducerRecord(topics[0], partition, System.currentTimeMillis(),
+                    UUID.randomUUID().toString(),
                     objectMapper.writeValueAsBytes(data))
             kafkaProducer.send(producerRecord).get()
             delay(5000)

doc/asciidoc/producer/index.adoc

@@ -10,15 +10,62 @@ include::configuration.adoc[]
 
 include::patterns.adoc[]
 
-=== Events
+=== Procedures
+
+The producer comes out with a list of procedures.
+Because of they use internal Neo4j API you must allow them with the following configuration paramater
+
+ dbms.security.procedures.unrestricted=streams.*
+
+If you are using them via Docker, pass this simple environment parameter:
+
+ NEO4J_dbms_security_procedures_unrestricted: streams.*
+
+
+==== streams.publish
+
+This procedure allows custom message streaming from Neo4j to the configured environment
+
+Uses:
+
+`CALL streams.publish('my-topic', 'Hello World from Neo4j!')`
+
+The message retrieved from the Consumer is the following:
+
+`{"payload":"Hello world from Neo4j!"}`
+
+Input Parameters:
+
+[cols="3*",options="header"]
+|===
+|Variable Name
+|Type
+|Description
+|`topic`
+|String
+|The topic where you want to publish the data
+
+|`payload`
+|Object
+|The data that you want to stream
+
+|===
+
+You can send any kind of data in the payload, nodes, relationships, paths, lists, maps, scalar values and nested versions thereof.
+
+=== Transaction Event Handler
+
+The transaction event handler is the core of the Stream Producer and allows to stream database changes.
+
+==== Events
 
 The Producer streams three kind of events:
 
 * *created*: when a node/relation/property is created
 * *updated*: when a node/relation/property is updated
 * *deleted*: when a node/relation/property is deleted
 
-==== Created
+===== Created
 
 Following an example of the node creation event:
 
@@ -30,7 +77,7 @@ include::data/node.created.json[]
 include::data/relationship.created.json[]
 ```
 
-==== Updated
+===== Updated
 
 Following an example of the node update event:
 
@@ -42,7 +89,7 @@ include::data/node.updated.json[]
 include::data/relationship.updated.json[]
 ```
 
-==== Deleted
+===== Deleted
 
 Following an example of the node creation event:
 
@@ -54,7 +101,7 @@ include::data/node.deleted.json[]
 include::data/relationship.deleted.json[]
 ```
 
-=== Meta
+==== Meta
 
 The *meta* field contains the metadata related to the transaction event:
 
@@ -93,7 +140,7 @@ The *meta* field contains the metadata related to the transaction event:
 |Contains the information about the source
 |===
 
-==== Source
+===== Source
 
 [cols="3*",options="header"]
 |===
@@ -107,7 +154,7 @@ The *meta* field contains the metadata related to the transaction event:
 |===
 
 
-=== Payload
+==== Payload
 
 The *payload* field contains the information about the the data related to the event
 
@@ -134,11 +181,11 @@ The *payload* field contains the information about the the data related to the e
 |The data after the transaction event
 |===
 
-==== Payload: before and after
+===== Payload: before and after
 
 We must distinguish two cases:
 
-===== Nodes
+====== Nodes
 
 [cols="3*",options="header"]
 |===
@@ -155,7 +202,7 @@ We must distinguish two cases:
 |List of properties attached to the node, the Key is the property name
 |===
 
-===== Relationships
+====== Relationships
 
 [cols="3*",options="header"]
 |===
@@ -198,7 +245,7 @@ We must distinguish two cases:
 |===
 
 
-=== Schema
+==== Schema
 
 [cols="3*",options="header"]
 |===
@@ -215,7 +262,7 @@ We must distinguish two cases:
 |The schema after the transaction event
 |===
 
-==== Schema: before and after
+===== Schema: before and after
 
 [cols="3*",options="header"]
 |===
@@ -234,7 +281,7 @@ We must distinguish two cases:
 
 We must distinguish two cases:
 
-===== Nodes
+====== Nodes
 
 [cols="3*",options="header"]
 |===
@@ -253,8 +300,9 @@ We must distinguish two cases:
 
 ====== Constraints
 
-A node can have a list of constraints attached to it:
+Nodes and Relationships can have a list of constraints attached to them:
 
+.Node Constraints
 [cols="3*",options="header"]
 |===
 |Field
@@ -274,8 +322,7 @@ A node can have a list of constraints attached to it:
 |List of node properties involved in the constraint
 |===
 
-===== Relationships
-
+.Relationship constraints
 [cols="3*",options="header"]
 |===
 |Field

performance/README.md

@@ -45,7 +45,7 @@ The  creation query used by the tester:
 The acquisition query:
 
 ```properties
-streams.sink.topic.cypher.neo4j=WITH event.value.payload AS payload, event.value.meta AS meta CALL apoc.do.case( [\
+streams.sink.topic.cypher.neo4j=WITH event.payload AS payload, event.meta AS meta CALL apoc.do.case( [\
 
 payload.type = 'node' AND meta.operation = 'created', \
 
@@ -100,10 +100,10 @@ nodes = 1000
 To check the coniguration or to get a fast result you can run the command
 
 ```shell
-python3.6 neo4j-streams-pt.py --start
+python3 neo4j-streams-pt.py --start
 ```
 
-The output is a windows that show you the distribution of the test. See result session to better understand the values. You can use the option `--plot-out file.png` to not show the result but save it on file. The details of the execution are dumped on standard output as CSV or you can redirect it on file using the option `--csv-file file.csv`
+The output is a windows that show you the distribution of the test. See result session to better understand the values. You can use the option `--plot-out file.png` to not show the result but save it on file. The details of the execution are dumped on standard output as CSV or you can redirect it on file using the option `--csv-out file.csv`
 
 ### baseline
 
@@ -114,13 +114,13 @@ ES: 3 = 3 * 1000 if 1000 is the value configured as *nodes* in the *[unit]* part
 To get the same result of *start*:
 
 ```shell
-python3.6 neo4j-streams-pt.py --baseline 1
+python3 neo4j-streams-pt.py --baseline 1
 ```
 
 The get more distribution (real case):
 
 ```shell
-python3.6 neo4j-streams-pt.py --baseline 1 10 100 1000 --plot-out results.png
+python3 neo4j-streams-pt.py --baseline 1 10 100 1000 --plot-out results.png
 ```
 
 This means the tester execute 5 (if *repeat = 5*) times each series. A series is comped by 1k, 10k, 100k, 1000k nodes (if *nodes = 1000*)

performance/docker-compose.yml

@@ -12,8 +12,8 @@ services:
       - "8474:7474"
       - "8687:7687"
     volumes:
-      - $HOME/app/neo4-streams/neo4j-community-3.4.7-producer/data:/data
-      - $HOME/app/neo4-streams/neo4j-community-3.4.7-producer/plugins:/plugins
+      - $HOME/app/neo4j-streams/neo4j-community-3.4.7-producer/data:/data
+      - $HOME/app/neo4j-streams/neo4j-community-3.4.7-producer/plugins:/plugins
     environment:
       NEO4J_kafka_zookeeper_connect: zookeeper:2181
       NEO4J_kafka_bootstrap_servers: broker:9092
@@ -32,15 +32,15 @@ services:
       - "7474:7474"
       - "7687:7687"
     volumes:
-      - $HOME/app/neo4-streams/neo4j-community-3.4.7-consumer/data:/data
-      - $HOME/app/neo4-streams/neo4j-community-3.4.7-consumer/plugins:/plugins
+      - $HOME/app/neo4j-streams/neo4j-community-3.4.7-consumer/data:/data
+      - $HOME/app/neo4j-streams/neo4j-community-3.4.7-consumer/plugins:/plugins
     environment:
       NEO4J_kafka_zookeeper_connect: zookeeper:2181
       NEO4J_kafka_bootstrap_servers: broker:9092
       NEO4J_AUTH: neo4j/consumer
       NEO4J_dbms_memory_heap_max__size: 2G
       NEO4J_kafka_max_poll_records: 16384
-      NEO4J_streams_sink_topic_cypher_neo4j: "WITH event.value.payload AS payload, event.value.meta AS meta CALL apoc.do.case( [
+      NEO4J_streams_sink_topic_cypher_neo4j: "WITH event.payload AS payload, event.meta AS meta CALL apoc.do.case( [
       payload.type = 'node' AND meta.operation = 'created', \
       'CREATE (x:Performance {received_time: apoc.date.currentTimestamp()}) SET x+=props RETURN count(x)'] 
       ,'RETURN 0', 

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

@@ -0,0 +1,19 @@
+package streams
+
+import org.neo4j.graphdb.Node
+import org.neo4j.graphdb.Relationship
+import streams.events.EntityType
+import streams.events.RelationshipNodeChange
+
+fun Map<String,String>.getInt(name:String, defaultValue: Int) = this.get(name)?.toInt() ?: defaultValue
+
+fun Node.toMap(): Map<String, Any?> {
+    return mapOf("id" to id.toString(), "properties" to allProperties, "labels" to labelNames(), "type" to EntityType.node)
+}
+
+fun Relationship.toMap(): Map<String, Any?> {
+    return mapOf("id" to id.toString(), "properties" to allProperties, "label" to type,
+            "start" to RelationshipNodeChange(startNode.id.toString(), startNode.labelNames()),
+            "end" to RelationshipNodeChange(endNode.id.toString(), endNode.labelNames()),
+            "type" to EntityType.relationship)
+}