Merge branch 'streambuilder-state-store' into develop-publish-1.5.0.100

Author: Plamen Pilev

core/Stream/IKStream.cs

@@ -1920,12 +1920,12 @@ void ForeachAsync(
             Func<ExternalRecord<K, V>, ExternalContext, Task> asyncAction,
             RetryPolicy retryPolicy = null,
             RequestSerDes<K, V> requestSerDes = null,
-            string named = null);
-        
+            string named = null);
+
         #endregion
-
+
         #region Process
-
+
         /// <summary>
         /// Process all records in this stream, one record at a time, by applying a processor (provided by the given
         /// <see cref="ProcessorSupplier{K,V}"/>.
@@ -1963,12 +1963,13 @@ void ForeachAsync(
         /// </summary>
         /// <param name="processorSupplier">an instance of <see cref="ProcessorSupplier{K,V}"/> which contains the processor and a potential state store. Use <see cref="ProcessorBuilder"/> to build this supplier.</param>
         /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
-        void Process(ProcessorSupplier<K, V> processorSupplier, string named = null);
-        
+        /// <param name="storeNames">The names of the state stores used by the processor.</param>
+        void Process(ProcessorSupplier<K, V> processorSupplier, string named = null, params string[] storeNames);
+
         #endregion
-
+
         #region Transform
-
+
         /// <summary>
         /// Transform each record of the input stream into zero or one record in the output stream (both key and value type
         /// can be altered arbitrarily).
@@ -2009,11 +2010,12 @@ void ForeachAsync(
         /// </summary>
         /// <param name="transformerSupplier">an instance of <see cref="TransformerSupplier{K,V,K1,V1}"/> which contains the transformer</param>
         /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        /// <param name="storeNames">The names of the state stores used by the processor.</param>
         /// <typeparam name="K1">the key type of the new stream</typeparam>
         /// <typeparam name="V1">the value type of the new stream</typeparam>
         /// <returns>a <see cref="IKStream{K1,V1}"/> that contains more or less records with new key and value (possibly of different type)</returns>
-        IKStream<K1, V1> Transform<K1, V1>(TransformerSupplier<K, V, K1, V1> transformerSupplier, string named = null);
-        
+        IKStream<K1, V1> Transform<K1, V1>(TransformerSupplier<K, V, K1, V1> transformerSupplier, string named = null, params string[] storeNames);
+
         /// <summary>
         /// Transform each record of the input stream into zero or one record in the output stream (only value type, the original key will be through to the upstream processors ).
         /// A <see cref="Transform{K,V1}"/> (provided by the given <see cref="TransformerSupplier{K,V,K,V1}"/>) is applied to each input record and
@@ -2053,9 +2055,10 @@ void ForeachAsync(
         /// </summary>
         /// <param name="transformerSupplier">an instance of <see cref="TransformerSupplier{K,V,K,V1}"/> which contains the transformer</param>
         /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        /// <param name="storeNames">The names of the state stores used by the processor.</param>
         /// <typeparam name="V1">the value type of the new stream</typeparam>
         /// <returns>a <see cref="IKStream{K,V1}"/> that contains more or less records with new key and value (possibly of different type)</returns>
-        IKStream<K, V1> TransformValues<V1>(TransformerSupplier<K, V, K, V1> transformerSupplier, string named = null);
+        IKStream<K, V1> TransformValues<V1>(TransformerSupplier<K, V, K, V1> transformerSupplier, string named = null, params string[] storeNames);
 
         #endregion
 

core/Stream/Internal/Graph/Nodes/StateStoreNode.cs

@@ -0,0 +1,20 @@
+using Streamiz.Kafka.Net.Processors.Internal;
+using Streamiz.Kafka.Net.State;
+
+namespace Streamiz.Kafka.Net.Stream.Internal.Graph.Nodes
+{
+    internal class StateStoreNode : StreamGraphNode
+    {
+        private readonly IStoreBuilder storeBuilder;
+
+        public StateStoreNode(IStoreBuilder storeBuilder, string streamGraphNode) : base(streamGraphNode)
+        {
+            this.storeBuilder = storeBuilder;
+        }
+
+        public override void WriteToTopology(InternalTopologyBuilder builder)
+        {
+            builder.AddStateStore(storeBuilder);
+        }
+    }
+}

core/Stream/Internal/Graph/Nodes/StatefulProcessorNode.cs

@@ -1,25 +1,26 @@
-using Streamiz.Kafka.Net.Processors;
-using Streamiz.Kafka.Net.Processors.Internal;
+using Streamiz.Kafka.Net.Processors.Internal;
 using Streamiz.Kafka.Net.State;
 
 namespace Streamiz.Kafka.Net.Stream.Internal.Graph.Nodes
 {
     internal class StatefulProcessorNode<K, V> : ProcessorGraphNode<K, V>
     {
         private readonly string[] storeNames;
-        private readonly IStoreBuilder storeBuilder;
-
+        private readonly IStoreBuilder storeBuilder;
+
         /// <summary>
         /// Create a node representing a stateful processor,
         /// where the store needs to be built and registered as part of building this node.
         /// </summary>
         /// <param name="nameNode"></param>
         /// <param name="parameters"></param>
         /// <param name="storeBuilder"></param>
-        public StatefulProcessorNode(string nameNode, ProcessorParameters<K, V> parameters, IStoreBuilder storeBuilder)
+        /// <param name="storeNames">The names of the state stores used by the processor.</param>
+
+        public StatefulProcessorNode(string nameNode, ProcessorParameters<K, V> parameters, IStoreBuilder storeBuilder, params string[] storeNames)
             : base(nameNode, parameters)
         {
-            storeNames = null;
+            this.storeNames = storeNames;
             this.storeBuilder = storeBuilder;
         }
 

core/Stream/Internal/InternalStreamBuilder.cs

@@ -137,9 +137,19 @@ internal void AddGraphNode(StreamGraphNode root, StreamGraphNode node)
             logger.LogDebug("Adding node {Node} in root node {Root}", node, root);
             root.AppendChild(node);
             nodes.Add(node);
-        }
-
-
+        }
+
+
         #endregion
+
+        #region Build Store
+        public void AddStateStore(IStoreBuilder storeBuilder)
+        {
+            var name = NewStoreName(string.Empty);
+
+            var node = new StateStoreNode(storeBuilder, name);
+            this.AddGraphNode(root, node);
+        }
+        #endregion
     }
 }

core/Stream/Internal/KStream.cs

@@ -104,12 +104,12 @@ public IKStream<K, V> FilterNot(Func<K, V, bool> predicate, string named = null)
 
         #region Process
 
-        public void Process(ProcessorSupplier<K, V> processorSupplier, string named = null)
+        public void Process(ProcessorSupplier<K, V> processorSupplier, string named = null, params string[] storeNames)
         {
             string name = new Named(named).OrElseGenerateWithPrefix(builder, KStream.PROCESSOR_NAME);
             ProcessorParameters<K, V> processorParameters = new ProcessorParameters<K, V>(
                 new KStreamProcessorSupplier<K, V>(processorSupplier), name);
-            StatefulProcessorNode<K, V> processorNode = new StatefulProcessorNode<K, V>(name, processorParameters, processorSupplier.StoreBuilder);
+            StatefulProcessorNode<K, V> processorNode = new StatefulProcessorNode<K, V>(name, processorParameters, processorSupplier.StoreBuilder, storeNames);
 
             builder.AddGraphNode(Node, processorNode);
         }
@@ -118,19 +118,19 @@ public void Process(ProcessorSupplier<K, V> processorSupplier, string named = nu
 
         #region Transform
 
-        public IKStream<K1, V1> Transform<K1, V1>(TransformerSupplier<K, V, K1, V1> transformerSupplier, string named = null)
-            => DoTransform(transformerSupplier, true, named);
+        public IKStream<K1, V1> Transform<K1, V1>(TransformerSupplier<K, V, K1, V1> transformerSupplier, string named = null, params string[] storeNames)
+            => DoTransform(transformerSupplier, true, named, storeNames);
 
-        public IKStream<K, V1> TransformValues<V1>(TransformerSupplier<K, V, K, V1> transformerSupplier, string named = null)
-            => DoTransform(transformerSupplier, false, named);
+        public IKStream<K, V1> TransformValues<V1>(TransformerSupplier<K, V, K, V1> transformerSupplier, string named = null, params string[] storeNames)
+            => DoTransform(transformerSupplier, false, named, storeNames);
 
         private IKStream<K1, V1> DoTransform<K1, V1>(TransformerSupplier<K, V, K1, V1> transformerSupplier, bool changeKey, string named =
-            null)
+            null, params string[] storeNames)
         {
             string name = new Named(named).OrElseGenerateWithPrefix(builder, changeKey ? KStream.TRANSFORM_NAME : KStream.TRANSFORMVALUES_NAME );
             ProcessorParameters<K, V> processorParameters = new ProcessorParameters<K, V>(
                 new KStreamTransformerSupplier<K, V, K1, V1>(transformerSupplier, changeKey), name);
-            StatefulProcessorNode<K, V> processorNode = new StatefulProcessorNode<K, V>(name, processorParameters, transformerSupplier.StoreBuilder);
+            StatefulProcessorNode<K, V> processorNode = new StatefulProcessorNode<K, V>(name, processorParameters, transformerSupplier.StoreBuilder, storeNames);
 
             builder.AddGraphNode(Node, processorNode);
 

core/StreamBuilder.cs

@@ -730,13 +730,29 @@ public IGlobalKTable<K, V> GlobalTable<K, V, KS, VS>(string topic, Materialized<
         public IGlobalKTable<K, V> GlobalTable<K, V, KS, VS>(string topic, Materialized<K, V, IKeyValueStore<Bytes, byte[]>> materialized, string named, ITimestampExtractor extractor)
             where KS : ISerDes<K>, new()
             where VS : ISerDes<V>, new()
-            => GlobalTable(topic, new KS(), new VS(), materialized, named, extractor);
-
-
+            => GlobalTable(topic, new KS(), new VS(), materialized, named, extractor);
+
+
         #endregion
-
+
         #endregion
+
+        #region State Store
+
+        /// <summary>
+        /// Adds a state store to the underlying <see cref="Topology"/>. 
+        /// It is required to connect state stores to <see cref="Streamiz.Kafka.Net.Processors.Public.IProcessor{K, V}"/> 
+        /// or <see cref="Streamiz.Kafka.Net.Processors.Public.ITransformer{K, V, K1, V1}"/>
+        /// before they can be used.
+        /// </summary>
+        /// <param name="storeBuilder">The builder used to obtain the <see cref="IStateStore"/> instance.</param>
+        public void AddStateStore(IStoreBuilder storeBuilder)
+        {
+            internalStreamBuilder.AddStateStore(storeBuilder);
+        }
 
+        #endregion
+
         /// <summary>
         /// Returns the <see cref="Topology"/> that represents the specified processing logic.
         /// Note that using this method means no optimizations are performed.