diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala b/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala index d9a087db7..7a673ac9a 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala @@ -7,7 +7,6 @@ package com.lightbend.kafka.scala.streams import org.apache.kafka.streams.kstream._ import org.apache.kafka.streams.{Consumed, KeyValue} import org.apache.kafka.common.serialization.Serde - import scala.language.implicitConversions /** diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedStreamS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedStreamS.scala index 57551f9ff..d3f9dc1ee 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedStreamS.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedStreamS.scala @@ -7,7 +7,6 @@ package com.lightbend.kafka.scala.streams import org.apache.kafka.streams.kstream._ import org.apache.kafka.streams.state.KeyValueStore import org.apache.kafka.common.utils.Bytes -import org.apache.kafka.common.serialization.Serde import ImplicitConversions._ import FunctionConversions._ @@ -16,19 +15,10 @@ import FunctionConversions._ */ class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) { - def count(): KTableS[K, Long] = { - val c: KTableS[K, java.lang.Long] = inner.count() - c.mapValues[Long](Long2long _) - } + def count(): KTableS[K, Long] = inner.count().asInstanceOf[KTable[K, Long]] - def count(store: String, keySerde: Option[Serde[K]] = None): KTableS[K, Long] = { - val materialized = keySerde.foldLeft(Materialized.as[K, java.lang.Long, KeyValueStore[Bytes, Array[Byte]]](store))( - (m, serde) => m.withKeySerde(serde) - ) - - val c: KTableS[K, java.lang.Long] = inner.count(materialized) - c.mapValues[Long](Long2long _) - } + def count(materialized: Materialized[K, Long, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, Long] = + inner.count(materialized) def reduce(reducer: (V, V) => V): KTableS[K, V] = inner.reduce((v1, v2) => reducer(v1, v2)) @@ -38,18 +28,6 @@ class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) { // works perfectly with Scala 2.12 though inner.reduce(((v1: V, v2: V) => reducer(v1, v2)).asReducer, materialized) - def reduce(reducer: (V, V) => V, storeName: String)(implicit keySerde: Serde[K], - valueSerde: Serde[V]): KTableS[K, V] = - // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place - // works perfectly with Scala 2.12 though - inner.reduce( - ((v1: V, v2: V) => reducer(v1, v2)).asReducer, - Materialized - .as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName) - .withKeySerde(keySerde) - .withValueSerde(valueSerde) - ) - def aggregate[VR](initializer: () => VR, aggregator: (K, V, VR) => VR): KTableS[K, VR] = inner.aggregate(initializer.asInitializer, aggregator.asAggregator) diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedTableS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedTableS.scala index 08acabd71..c1e512d23 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedTableS.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/KGroupedTableS.scala @@ -15,14 +15,9 @@ import FunctionConversions._ */ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) { - type ByteArrayKVStore = KeyValueStore[Bytes, Array[Byte]] + def count(): KTableS[K, Long] = inner.count().asInstanceOf[KTable[K, Long]] - def count(): KTableS[K, Long] = { - val c: KTableS[K, java.lang.Long] = inner.count() - c.mapValues[Long](Long2long(_)) - } - - def count(materialized: Materialized[K, Long, ByteArrayKVStore]): KTableS[K, Long] = + def count(materialized: Materialized[K, Long, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, Long] = inner.count(materialized) def reduce(adder: (V, V) => V, subTractor: (V, V) => V): KTableS[K, V] = @@ -32,7 +27,7 @@ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) { def reduce(adder: (V, V) => V, subtractor: (V, V) => V, - materialized: Materialized[K, V, ByteArrayKVStore]): KTableS[K, V] = + materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place // works perfectly with Scala 2.12 though inner.reduce(((v1, v2) => adder(v1, v2)).asReducer, ((v1, v2) => subtractor(v1, v2)).asReducer, materialized) @@ -43,6 +38,6 @@ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) { def aggregate[VR](initializer: () => VR, adder: (K, V, VR) => VR, subtractor: (K, V, VR) => VR, - materialized: Materialized[K, VR, ByteArrayKVStore]): KTableS[K, VR] = + materialized: Materialized[K, VR, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, VR] = inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator, materialized) } diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/Materialized.scala b/src/main/scala/com/lightbend/kafka/scala/streams/Materialized.scala new file mode 100644 index 000000000..32d1b00c6 --- /dev/null +++ b/src/main/scala/com/lightbend/kafka/scala/streams/Materialized.scala @@ -0,0 +1,29 @@ +package com.lightbend.kafka.scala.streams + +import org.apache.kafka.common.serialization.Serde +import org.apache.kafka.common.utils.Bytes +import org.apache.kafka.streams.kstream.{Materialized => JavaMaterialized} +import org.apache.kafka.streams.processor.StateStore +import org.apache.kafka.streams.state._ + +object Materialized { + + def as[K, V, S <: StateStore](storeName: String)(implicit keySerde: Serde[K], + valueSerde: Serde[V]): JavaMaterialized[K, V, S] = + JavaMaterialized.as(storeName).withKeySerde(keySerde).withValueSerde(valueSerde) + + def as[K, V]( + supplier: WindowBytesStoreSupplier + )(implicit keySerde: Serde[K], valueSerde: Serde[V]): JavaMaterialized[K, V, WindowStore[Bytes, Array[Byte]]] = + JavaMaterialized.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde) + + def as[K, V]( + supplier: SessionBytesStoreSupplier + )(implicit keySerde: Serde[K], valueSerde: Serde[V]): JavaMaterialized[K, V, SessionStore[Bytes, Array[Byte]]] = + JavaMaterialized.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde) + + def as[K, V]( + supplier: KeyValueBytesStoreSupplier + )(implicit keySerde: Serde[K], valueSerde: Serde[V]): JavaMaterialized[K, V, KeyValueStore[Bytes, Array[Byte]]] = + JavaMaterialized.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde) +} diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/SessionWindowedKStreamS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/SessionWindowedKStreamS.scala index 39475a83f..65535ab3b 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/SessionWindowedKStreamS.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/SessionWindowedKStreamS.scala @@ -8,7 +8,6 @@ import org.apache.kafka.streams.kstream._ import org.apache.kafka.streams.state.SessionStore import org.apache.kafka.common.utils.Bytes import FunctionConversions._ - import ImplicitConversions._ /** @@ -27,10 +26,7 @@ class SessionWindowedKStreamS[K, V](val inner: SessionWindowedKStream[K, V]) { materialized: Materialized[K, VR, SessionStore[Bytes, Array[Byte]]]): KTableS[Windowed[K], VR] = inner.aggregate(initializer.asInitializer, aggregator.asAggregator, merger.asMerger, materialized) - def count(): KTableS[Windowed[K], Long] = { - val c: KTableS[Windowed[K], java.lang.Long] = inner.count() - c.mapValues[Long](Long2long(_)) - } + def count(): KTableS[K, Long] = inner.count().asInstanceOf[KTable[K, Long]] def count(materialized: Materialized[K, Long, SessionStore[Bytes, Array[Byte]]]): KTableS[Windowed[K], Long] = inner.count(materialized) diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/StreamsBuilderS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/StreamsBuilderS.scala index e5b342761..128edc622 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/StreamsBuilderS.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/StreamsBuilderS.scala @@ -8,7 +8,7 @@ import java.util.regex.Pattern import com.lightbend.kafka.scala.streams.ImplicitConversions._ import org.apache.kafka.common.utils.Bytes -import org.apache.kafka.streams.kstream.{GlobalKTable, Materialized} +import org.apache.kafka.streams.kstream.{GlobalKTable, Materialized => JavaMaterialized} import org.apache.kafka.streams.processor.{ProcessorSupplier, StateStore} import org.apache.kafka.streams.state.{KeyValueStore, StoreBuilder} import org.apache.kafka.streams.{Consumed, StreamsBuilder, Topology} @@ -32,7 +32,7 @@ class StreamsBuilderS(inner: StreamsBuilder = new StreamsBuilder) { def table[K, V](topic: String)(implicit consumed: Consumed[K, V]): KTableS[K, V] = inner.table[K, V](topic, consumed) - def table[K, V](topic: String, materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]])( + def table[K, V](topic: String, materialized: JavaMaterialized[K, V, KeyValueStore[Bytes, Array[Byte]]])( implicit consumed: Consumed[K, V] ): KTableS[K, V] = inner.table[K, V](topic, consumed, materialized) @@ -40,7 +40,7 @@ class StreamsBuilderS(inner: StreamsBuilder = new StreamsBuilder) { def globalTable[K, V](topic: String)(implicit consumed: Consumed[K, V]): GlobalKTable[K, V] = inner.globalTable(topic, consumed) - def globalTable[K, V](topic: String, materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]])( + def globalTable[K, V](topic: String, materialized: JavaMaterialized[K, V, KeyValueStore[Bytes, Array[Byte]]])( implicit consumed: Consumed[K, V] ): GlobalKTable[K, V] = inner.globalTable(topic, consumed, materialized) diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/TimeWindowedKStreamS.scala b/src/main/scala/com/lightbend/kafka/scala/streams/TimeWindowedKStreamS.scala index ee484f6c7..e4291c7af 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/TimeWindowedKStreamS.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/TimeWindowedKStreamS.scala @@ -7,7 +7,6 @@ package com.lightbend.kafka.scala.streams import org.apache.kafka.streams.kstream._ import org.apache.kafka.streams.state.WindowStore import org.apache.kafka.common.utils.Bytes -import org.apache.kafka.common.serialization.Serde import ImplicitConversions._ import FunctionConversions._ @@ -24,19 +23,12 @@ class TimeWindowedKStreamS[K, V](val inner: TimeWindowedKStream[K, V]) { materialized: Materialized[K, VR, WindowStore[Bytes, Array[Byte]]]): KTableS[Windowed[K], VR] = inner.aggregate(initializer.asInitializer, aggregator.asAggregator, materialized) - def count(): KTableS[Windowed[K], Long] = { - val c: KTableS[Windowed[K], java.lang.Long] = inner.count() - c.mapValues[Long](Long2long(_)) - } - - def count(store: String, keySerde: Option[Serde[K]] = None): KTableS[Windowed[K], Long] = { - val materialized = { - val m = Materialized.as[K, java.lang.Long, WindowStore[Bytes, Array[Byte]]](store) - keySerde.foldLeft(m)((m, serde) => m.withKeySerde(serde)) - } - val c: KTableS[Windowed[K], java.lang.Long] = inner.count(materialized) - c.mapValues[Long](Long2long(_)) - } + def count(): KTableS[Windowed[K], Long] = inner.count().asInstanceOf[KTable[Windowed[K], Long]] + + def count( + materialized: Materialized[Windowed[K], Long, WindowStore[Bytes, Array[Byte]]] + ): KTableS[Windowed[K], Long] = + inner.count(materialized) def reduce(reducer: (V, V) => V): KTableS[Windowed[K], V] = inner.reduce(reducer.asReducer)