Merge pull request #123 from compstak/debezium-table-expose-filter

Expose filter operation for DebeziumTable
compstak · Feb 10, 2021 · 91918f4 · 91918f4
2 parents 9e6b439 + cc33f23
commit 91918f4
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/.gitignore b/.gitignore
@@ -7,6 +7,7 @@
 .metals/
 .vscode
 metals.sbt
+.bsp/
 
 .DS_Store
 
@@ -22,4 +23,4 @@ target/
 
 test-reports
 
-.env
+.env
diff --git a/debezium/src/main/scala/compstak/kafkastreams4s/debezium/DebeziumTable.scala b/debezium/src/main/scala/compstak/kafkastreams4s/debezium/DebeziumTable.scala
@@ -31,6 +31,9 @@ case class DebeziumTable[K: Encoder: Decoder, V: Encoder: Decoder](
   def mapFilterWithKey[V2: Encoder: Decoder](f: (DebeziumKey[K], V) => Option[V2]): DebeziumTable[K, V2] =
     copy(toCirceTable = toCirceTable.mapFilterWithKey(f))
 
+  def filter(f: V => Boolean): DebeziumTable[K, V] =
+    copy(toCirceTable = toCirceTable.filter(f))
+
   def join[K2, V2, Z: Encoder: Decoder](
     other: DebeziumTable[K2, V2]
   )(f: V => K2)(g: (V, V2) => Z): DebeziumTable[K, Z] =