Merge pull request #55 from mtranter/ktable-recursion-fix

debasishg · web-flow · commit 13579e526e70 · 2018-03-05T11:15:06.000+05:30
Force (K,V)=&gt;Boolean to Predicate&lt;K,V&gt; in KTableS
diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/KTableS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/KTableS.scala
@@ -21,7 +21,7 @@ class KTableS[K, V](val inner: KTable[K, V]) {
 
   def filter(predicate: (K, V) => Boolean,
     materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {
-    inner.filter(predicate(_, _), materialized)
+    inner.filter(predicate.asPredicate, materialized)
   }
 
   def filterNot(predicate: (K, V) => Boolean): KTableS[K, V] = {
@@ -30,7 +30,7 @@ class KTableS[K, V](val inner: KTable[K, V]) {
 
   def filterNot(predicate: (K, V) => Boolean,
     materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {
-    inner.filterNot(predicate(_, _), materialized)
+    inner.filterNot(predicate.asPredicate, materialized)
   }
 
   def mapValues[VR](mapper: V => VR): KTableS[K, VR] = {

Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ class KTableS[K, V](val inner: KTable[K, V]) {`
`21`	`21`
`22`	`22`	`def filter(predicate: (K, V) => Boolean,`
`23`	`23`	`materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {`
`24`		`- inner.filter(predicate(_, _), materialized)`
	`24`	`+ inner.filter(predicate.asPredicate, materialized)`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`def filterNot(predicate: (K, V) => Boolean): KTableS[K, V] = {`
`@@ -30,7 +30,7 @@ class KTableS[K, V](val inner: KTable[K, V]) {`
`30`	`30`
`31`	`31`	`def filterNot(predicate: (K, V) => Boolean,`
`32`	`32`	`materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {`
`33`		`- inner.filterNot(predicate(_, _), materialized)`
	`33`	`+ inner.filterNot(predicate.asPredicate, materialized)`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`def mapValues[VR](mapper: V => VR): KTableS[K, VR] = {`