extend Transform and TransformValues with state stores

Author: Plamen Pilev

core/Stream/IKStream.cs

@@ -1964,12 +1964,12 @@ void ForeachAsync(
         /// <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>
         /// <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);
-        
+        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).
@@ -2010,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
@@ -2054,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/KStream.cs

@@ -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);