lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

delete extensions, they were moved to kotka

Browse source
This commit is contained in:
Adam 2022-01-21 07:57:07 +01:00
parent 27776553ba
commit 8c107b7b84
12 changed files with 0 additions and 639 deletions
Download patch file Download diff file Expand all files Collapse all files

31
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt
View file

@ -1,31 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.streams.state.QueryableStoreTypes.timestampedKeyValueStore
///**
// * * update topic [topicName]
// * * plus table-source topic [tableName]
// * * plus state store [globalStoreName]
// *
// * equals a GlobalTableDefinition
// */
//abstract class GlobalKTableDefinition<K, V>(
// val tableName: String,
// override val serdes: KeyValueSerdes<K, V>,
// val globalStoreName: String = "$tableName-store",
// /**
// * The backing topic, from which this table will be filled.
// *
// * Updates to items can be sent or received with this topic.
// */
// final override val topicName: String = "$tableName-topic",
// val storeType: TimestampedQueryStoreType<K, V> = timestampedKeyValueStore(),
//) : TopicDefinition<K, V> {
// val topicNames: Set<String>
//
// init {
// require(tableName != topicName)
// require(tableName != globalStoreName)
// topicNames = setOf(tableName, topicName)
// }
//}

32
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt
View file

@ -1,32 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import org.apache.kafka.common.utils.Bytes
//import org.apache.kafka.streams.kstream.KGroupedStream
//import org.apache.kafka.streams.kstream.KTable
//import org.apache.kafka.streams.kstream.Materialized
//import org.apache.kafka.streams.state.KeyValueStore
//
//
//fun <K> KGroupedStream<K, *>.count(
// name: String? = null,
// materialized: Materialized<K, Long, KeyValueStore<Bytes, ByteArray>>? = null,
//): KTable<K, Long> {
// return when {
// name != null && materialized != null -> count(withName(name), materialized)
// name == null && materialized != null -> count(materialized)
// name != null && materialized == null -> count(withName(name))
// else -> count()
// }
//}
//
//
//
////<VOut> CogroupedKStream<K, VOut> cogroup(final Aggregator<? super K, ? super V, VOut> aggregator);
////<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,final Aggregator<? super K, ? super V, VR> aggregator, final Named named, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<W extends Window> TimeWindowedKStream<K, V> windowedBy(final Windows<W> windows);
////KTable<K, Long> count(final Materialized<K, Long, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> reduce(final Reducer<V> reducer);
////KTable<K, V> reduce(final Reducer<V> reducer, final Named named, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> reduce(final Reducer<V> reducer,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////SessionWindowedKStream<K, V> windowedBy(final SessionWindows windows);
////TimeWindowedKStream<K, V> windowedBy(final SlidingWindows windows);

152
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt
View file

@ -1,152 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.KGroupedStream
//import org.apache.kafka.streams.kstream.KStream
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.processor.TopicNameExtractor
//
//// /** @see KStream.map */
//// inline fun <inK, inV, reified outK, reified outV : TopicData<outK>> KStream<inK, inV>.map(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> outV
//// ): KStream<outK, outV> =
//// map({ k, v -> mapper(k, v).asKeyValue() }, withName(name))
//
///** @see KStream.map */
//inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.map(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Pair<outK, outV>
//): KStream<outK, outV> =
// map({ k, v -> mapper(k, v).toKeyValue() }, withName(name))
//
///** @see [KStream.mapValues] */
//inline fun <K, inV, outV> KStream<K, inV>.mapValues(
// name: String,
// crossinline mapper: (key: K, value: inV) -> outV
//): KStream<K, outV> = mapValues({ k, v -> mapper(k, v) }, withName(name))
//
//// /** @see KStream.flatMap */
//// inline fun <inK, reified outK, inV, reified outV : TopicData<outK>> KStream<inK, inV>.flatMap(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> Iterable<outV>
//// ): KStream<outK, outV> = flatMap(
//// { k, v -> mapper(k, v).map { a: outV -> a.toKeyValue() } },
//// withName(name)
//// )
//
///** @see KStream.flatMap */
//inline fun <reified inK, inV, reified outK, reified outV> KStream<inK, inV>.flatMap(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
//): KStream<outK, outV> = flatMap(
// { k, v -> mapper(k, v).map { it.toKeyValue() } },
// withName(name)
//)
//
///** @see KStream.groupBy */
//fun <K, V, outK> KStream<K, V>.groupBy(
// grouped: Grouped<outK, V>,
// keySelector: (K, V) -> outK
//): KGroupedStream<outK, V> = groupBy(keySelector, grouped)
//
//fun <K, V> KStream<K, V>.to(
// produced: Produced<K, V>,
// topicNameExtractor: TopicNameExtractor<K, V>,
//) = to(topicNameExtractor, produced)
//
//fun <K, V> KStream<K, V>.filter(
// name: String,
// predicate: (K, V) -> Boolean,
//): KStream<K, V> = filter(predicate, withName(name))
//
//
//fun <K, V> KStream<K, V>.filterNot(
// name: String,
// predicate: (K, V) -> Boolean,
//): KStream<K, V> = filterNot(predicate, withName(name))
//
//
//fun <K, V> KStream<K, V>.merge(
// name: String,
// other: KStream<K, V>,
//): KStream<K, V> = merge(other, withName(name))
//
//
//// KStream<K, V> filter(final Predicate<? super K, ? super V> predicate, final Named named);
//// KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final Named named);
//
//// @JvmName("flatMapKeyValuePairs")
//// inline fun <inK, reified outK, inV, reified outV> KStream<inK, inV>.flatMap(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
//// ): KStream<outK, outV> =
//// flatMap(
//// { k, v -> mapper(k, v).map { it.asKeyPair() } },
//// withName(name)
//// )
//
//// inline fun <reified K, inV, reified outV> KStream<K, inV>.mapValues(
//// name: String,
//// crossinline mapper: (readOnlyKey: K, value: inV) -> outV
//// ): KStream<K, outV> =
//// map({ k, v -> (k to mapper(v)).asKeyPair() }, KafkaStreamsExtensions.withName(name))
//
//
////open class TransformerKt<inK, inV, outK, outV>(
////) : Transformer<inK, inV, KeyValue<outK, outV>> {
////
//// val onClose: TransformerKt<inK, inV, outK, outV>.() -> Unit = {}
//// val transformer: TransformerKt<inK, inV, outK, outV>.(key: inK, value: inV) -> KeyValue<outK, outV>? =
//// {}
////
//// private lateinit var _context: ProcessorContext
//// val context: ProcessorContext?
//// get() = if (this::_context.isInitialized) _context else null
////
//// override fun init(context: ProcessorContext) {
//// this._context = context
//// }
////
//// override fun close() {
//// onClose()
//// }
////
//// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
//// return transformer(key, value)
//// }
////}
////
////
//// /** one or the other or both */
//// KTable<K, V> toTable(final Named named, Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
//// <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalTable, KeyValueMapper<? super K, ? super V, ? extends GK> keySelector, ValueJoinerWithKey<? super K, ? super V, ? super GV, ? extends RV> joiner, Named named);
//// <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalTable, KeyValueMapper<? super K, ? super V, ? extends GK> keySelector, ValueJoinerWithKey<? super K, ? super V, ? super GV, ? extends RV> valueJoiner, Named named);
//// <K1, V1> KStream<K1, V1> flatTransform(final TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier, Named named, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> flatTransform(final TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> transform(final TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, Named named, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> transform(final TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames);
//// <KR, VR> KStream<KR, VR> flatMap(final KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends KR, ? extends VR>>> mapper, Named named);
//// <KR, VR> KStream<KR, VR> map(final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> mapper, Named named);
//// <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR> keySelector, Grouped<KR, V> grouped);
//// <KR> KStream<KR, V> selectKey(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper, Named named);
//// <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VR> KStream<K, VR> flatMapValues(final ValueMapperWithKey<? super K, ? super V, ? extends Iterable<? extends VR>> mapper, Named named);
//// <VR> KStream<K, VR> flatTransformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, Iterable<VR>> valueTransformerSupplier, Named named, String... stateStoreNames);
//// <VR> KStream<K, VR> mapValues(final ValueMapperWithKey<? super K, ? super V, ? extends VR> mapper, Named named);
//// <VR> KStream<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier, Named named, String... stateStoreNames);
//// <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, ValueJoinerWithKey<? super K, ? super V, ? super VT, ? extends VR> joiner, Joined<K, V, VT> joined);
//// <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table, ValueJoinerWithKey<? super K, ? super V, ? super VT, ? extends VR> joiner, Joined<K, V, VT> joined);
//// BranchedKStream<K, V> split(final Named named);
//// KGroupedStream<K, V> groupByKey(final Grouped<K, V> grouped);
//// KStream<K, V> merge(final KStream<K, V> stream, final Named named);
//// KStream<K, V> peek(final ForeachAction<? super K, ? super V> action, final Named named);
//// KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
//// void foreach(final ForeachAction<? super K, ? super V> action, final Named named);
//// void print(final Printed<K, V> printed);
//// void process(final ProcessorSupplier<? super K, ? super V, Void, Void> processorSupplier, Named named, String... stateStoreNames);
//// void to(final String topic, Produced<K, V> produced);
//// void to(final TopicNameExtractor<K, V> topicExtractor, Produced<K, V> produced);
////

90
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt
View file

@ -1,90 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import dev.adamko.kafkatorio.extensions.kafkastreams.*
//import kotlin.experimental.ExperimentalTypeInference
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.kstream.KStream
//import org.apache.kafka.streams.kstream.Transformer
//import org.apache.kafka.streams.kstream.TransformerSupplier
//import org.apache.kafka.streams.processor.ProcessorContext
//
///** @see KStream.transform */
//inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.transformKt(
// crossinline block: BuildTransformer<inK, inV, outK, outV>
//): KStream<outK, outV> {
//
// val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
//
// private lateinit var scope: TransformerScope<inK, inV, outK, outV>
//
// override fun init(procContext: ProcessorContext) {
// scope = TransformerScope(procContext)
// scope.block()
// }
//
// override fun close() {
// scope.close()
// }
//
// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
// return scope.mapper(key, value)?.toKeyValue()
// }
// }
//
// return transform({ transformer })
//}
//
///** @see KStream.transform */
//@OptIn(ExperimentalTypeInference::class)
//inline fun <inK, inV, reified outK, reified outV, outKV : KeyValue<outK, outV>> KStream<inK, inV>.transformKt2(
// @BuilderInference block: Transformer<inK, inV, outKV>.() -> Unit
//): KStream<outK, outV> {
//
// val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
//
// private lateinit var scope: TransformerScope<inK, inV, outK, outV>
//
// override fun init(procContext: ProcessorContext) {
// scope = TransformerScope(procContext)
// }
//
// override fun close() {
// scope.close()
//
// sequence<Long> {
//
// }
// }
//
// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
// return scope.mapper(key, value)?.toKeyValue()
// }
// }
//
// return transform(TransformerSupplier { transformer })
//}
//
//typealias BuildTransformer<inK, inV, outK, outV> = TransformerScope<inK, inV, outK, outV>.() -> Unit
//
//typealias TransformerMapper<inK, inV, outK, outV> = (key: inK, value: inV) -> Pair<outK, outV>?
//typealias TransformerCloseHandler = (ProcessorContext) -> Unit
//
//open class TransformerScope<inK, inV, outK, outV>(
// val processorContext: ProcessorContext,
//) : AutoCloseable {
//
// private var closeHandler: TransformerCloseHandler = {}
// lateinit var mapper: TransformerMapper<inK, inV, outK, outV>
//
// fun handleClose(closeHandler: TransformerCloseHandler) {
// this.closeHandler = closeHandler
// }
//
// fun mapEach(mapper: TransformerMapper<inK, inV, outK, outV>) {
// this.mapper = mapper
// }
//
// override fun close() {
// closeHandler(processorContext)
// }
//}

48
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt
View file

@ -1,48 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.KGroupedTable
//import org.apache.kafka.streams.kstream.KTable
//import org.apache.kafka.streams.kstream.KeyValueMapper
//
///** @see [KTable.mapValues] */
//fun <K, inV, outV> KTable<K, inV>.mapValues(
// name: String,
// mapper: (readOnlyKey: K, value: inV) -> outV
//): KTable<K, outV> =
// mapValues(
// { k, v -> mapper(k, v) },
// withName(name)
// )
//
///** @See [KTable.groupBy] */
//fun <inK, inV, outK, outV> KTable<inK, inV>.groupBy(
// grouped: Grouped<outK, outV>? = null,
// selector: KeyValueMapper<inK, inV, KeyValue<outK, outV>>
//): KGroupedTable<outK, outV> = when (grouped) {
// null -> groupBy(selector)
// else -> groupBy(selector, grouped)
//}
//
//
////
////<KR> KStream<KR, V> toStream(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper,final Named named);
////<VO, VR> KTable<K, VR> join(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR, KO, VO> KTable<K, VR> join(final KTable<KO, VO> other,final Function<V, KO> foreignKeyExtractor,final ValueJoiner<V, VO, VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////
////<VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////
////<VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR, KO, VO> KTable<K, VR> leftJoin(final KTable<KO, VO> other,final Function<V, KO> foreignKeyExtractor,final ValueJoiner<V, VO, VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> mapValues(final ValueMapper<? super V, ? extends VR> mapper,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> mapValues(final ValueMapperWithKey<? super K, ? super V, ? extends VR> mapper,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> transformerSupplier,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized,final Named named,final String... stateStoreNames);
////KStream<K, V> toStream(final Named named);
/////** materialized, named - optional */
////KTable<K, V> filter(final Predicate<? super K, ? super V> predicate,final Named named,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
/////** materialized, named - optional */
////KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate,final Named named,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> suppress(final Suppressed<? super K> suppressed);
////String queryableStoreName();

91
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt
View file

@ -1,91 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.Joined
//import org.apache.kafka.streams.kstream.Materialized
//import org.apache.kafka.streams.kstream.Named
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.kstream.Repartitioned
//import org.apache.kafka.streams.processor.StateStore
//
///** Helpful extensions & shortcuts for Kafka Streams */
//
//
///** @see [Named] */
//fun withName(name: String): Named = Named.`as`(name)
//
//// @Suppress("FunctionName")
//// inline fun <reified K, reified V, S : StateStore> MaterializedAs(
//// tableDefinition: GlobalTableDefinition<K, V>
//// ) = MaterializedAs<K, V, S>(
//// tableDefinition.globalStoreName,
//// tableDefinition.serdes.keySerde,
//// tableDefinition.serdes.valueSerde
//// )
//
///** @see Materialized */
//inline fun <reified K, reified V, S : StateStore> materializedAs(
// name: String,
// keySerde: Serde<K>,//= kxsJsonSerde(),
// valueSerde: Serde<V>,//= kxsJsonSerde(),
//): Materialized<K, V, S> =
// Materialized.`as`<K, V, S>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
///** @see Repartitioned */
//inline fun <reified Key, reified Val> repartitionedAs(
// name: String,
// keySerde: Serde<Key>,
// valueSerde: Serde<Val>,
//): Repartitioned<Key, Val> =
// Repartitioned.`as`<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
///** @see Produced */
//fun <K, V> producedAs(name: String): Produced<K, V> =
// Produced.`as`(name)
//
///** @see Produced */
//inline fun <reified Key, reified Val> producedAs(
// name: String,
// keySerde: Serde<Key>,
// valueSerde: Serde<Val>,
//): Produced<Key, Val> =
// producedAs<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
//// @Suppress("FunctionName")
//// fun <Key, Val, V_other> JoinedAs(name: String): Joined<Key, Val, V_other> =
//// Joined.`as`(name)
//
///** @see Joined */
//inline fun <reified Key, reified Val, reified V_other> joinedAs(
// name: String,
// keySerde: Serde<Key>,// = kxsJsonSerde(),
// valueSerde: Serde<Val>,//= kxsJsonSerde(),
// otherValueSerde: Serde<V_other>?,//= kxsJsonSerde(),
//): Joined<Key, Val, V_other> =
// Joined.`as`<Key, Val, V_other>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
// .withOtherValueSerde(otherValueSerde)
//
///** @see [Grouped.`as`] */
//inline fun <reified Key, reified Val> groupedAs(
// name: String,
// keySerde: Serde<Key>?, //= jsonMapper.serde(),
// valueSerde: Serde<Val>?, //= jsonMapper.serde(),
//): Grouped<Key, Val> =
// Grouped.`as`<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
//// fun <K, V> InteractiveQueryService.getQueryableStore(
//// tableDef: GlobalTableDefinition<K, V>
//// ): Optional<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> {
//// return getQueryableStore(tableDef.globalStoreName, tableDef.storeType)
//// }

33
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt
View file

@ -1,33 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.state.ValueAndTimestamp
//
//
//inline fun <reified K, reified V> Pair<K, V>.toKeyValue(): KeyValue<K, V> =
// KeyValue.pair(first, second)
//
//infix fun <K, V> K.kvPair(value: V): KeyValue<K, V> = KeyValue(this, value)
//
//infix fun <V> V.timestamped(timestamp: Long): ValueAndTimestamp<V> =
// ValueAndTimestamp.make(this, timestamp)
//
//
//@JvmName("keyValueGetKey")
//operator fun <K, V> KeyValue<K, V>.component1(): K = key
//@JvmName("keyValueGetValue")
//operator fun <K, V> KeyValue<K, V>.component2(): V = value
//
//
//@JvmName("keyValueTimestampGetKey")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component1(): K = key
//@JvmName("keyValueTimestampGetValue")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component2(): V = value.value()
//@JvmName("keyValueTimestampGetTimestamp")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component3(): Long = value.timestamp()
//
//
//@JvmName("valueTimestampGetValue")
//operator fun <V> ValueAndTimestamp<V>.component1(): V = value()
//@JvmName("valueTimestampGetTimestamp")
//operator fun <V> ValueAndTimestamp<V>.component2(): Long = timestamp()

41
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt
View file

@ -1,41 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//import dev.adamko.kafkatorio.extensions.kxs.KotlinxSerializationExtensions.serde
//import dev.adamko.kafkatorio.processor.jsonMapper
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.common.serialization.Serdes
//import org.apache.kafka.streams.kstream.Consumed
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.kstream.Repartitioned
//
///**
// * A wrapper for two [Serde]s - [one for the Key][keySerde], [another for the Value][valueSerde].
// */
//open class KeyValueSerdes<K, V>(
// val keySerde: Serde<K>,
// val valueSerde: Serde<V>,
//) {
//
// fun consumer(name: String): Consumed<K, V> =
// Consumed.with(keySerde, valueSerde).withName(name)
//
// fun producer(name: String): Produced<K, V> =
// Produced.with(keySerde, valueSerde).withName(name)
//
// fun repartitioned(name: String): Repartitioned<K, V> =
// Repartitioned.with(keySerde, valueSerde).withName(name)
//
// companion object {
// inline fun <reified K, reified V> kxsJson(): KeyValueSerdes<K, V> =
// KeyValueSerdes(
// jsonMapper.serde(),
// jsonMapper.serde(),
// )
//
// fun string(): KeyValueSerdes<String, String> =
// KeyValueSerdes<String, String>(
// Serdes.String(),
// Serdes.String(),
// )
// }
//}

17
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt
View file

@ -1,17 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//import org.apache.kafka.streams.KeyValue
//
//interface TopicData<K> {
//
// val topicKey: K
//
// companion object {
// fun <K, T : TopicData<K>> T.toPair(): Pair<K, T> = topicKey to this
// fun <K, T : TopicData<K>> T.toKeyValue(): KeyValue<K, T> = KeyValue.pair(topicKey, this)
//
// operator fun <K, T : TopicData<K>> T.component1(): K = topicKey
// operator fun <K, T : TopicData<K>> T.component2(): T = this
//
// }
//}

42
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt
View file

@ -1,42 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams;
///**
// * A definition for all topics that are created in Kafka.
// *
// * These can be used to create KTopics, KTables, or GlobalKTables.
// */
//interface TopicDefinition<K, V> {
//
// /** unique kafka processor id */
// val pid: String
// get() = this::class.simpleName!!
//
// /**
// * Unique name for each [TopicDefinition]
// *
// * https://devshawn.com/blog/apache-kafka-topic-naming-conventions/
// *
// * https://www.xeotek.com/topic-naming-conventions-how-do-i-name-my-topics-5-recommendations-with-examples/
// *
// * Domain: warehouse
// * Operation: stock-management
// * Data type:
// * - cdc=all instances of a specific object/thing/item,
// * - cmd=command (mutation),
// * - fct=fact (information (complete or partial)
// * about an event at specific point in time)
// * - sys=internal system topic - not meant for external use
// * - tbl=a global table store (must be paired with another topic)
// * Description: the data type
// *
// * Version: stored in message header
// *
// * ```
// * warehouse.stock-management.sts/cdc/cmd.warehouse-item
// * ```
// */
// val topicName: String
//
// val serdes: KeyValueSerdes<K, V>
//
//}

7
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt
View file

@ -1,7 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.streams.state.QueryableStoreType
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
import org.apache.kafka.streams.state.ValueAndTimestamp
//typealias TimestampedQueryStoreType<K, V> = QueryableStoreType<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>>

55
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt
View file

@ -1,55 +0,0 @@
package dev.adamko.kafkatorio.extensions.kxs
//import kotlinx.serialization.decodeFromString
//import kotlinx.serialization.encodeToString
//import kotlinx.serialization.json.Json
//import org.apache.kafka.common.serialization.Deserializer
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.common.serialization.Serializer
//
//object KotlinxSerializationExtensions {
//
// inline fun <reified T> Json.kafkaSerializer() = Serializer { topic: String, data: T ->
// runCatching {
// encodeToString(data).encodeToByteArray()
//// jsonMapper.encodeToString(data as T).encodeToByteArray()
// }.getOrElse { e ->
// println(
// """
// Exception on encodeToString
// Topic: $topic
// topicData: $data
// topicData as T: ${data as T}
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
// inline fun <reified T> Json.kafkaDeserializer() = Deserializer { topic: String, data: ByteArray ->
// runCatching {
// decodeFromString<T>(data.decodeToString())
// }.getOrElse { e ->
// println(
// """
// Exception on decodeFromString,
// Topic: $topic
// topicData: $data
// topicData as T: ${data as T}
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
//
// inline fun <reified T> Json.serde() = object : Serde<T> {
//
// override fun serializer(): Serializer<T> = kafkaSerializer()
//
// override fun deserializer(): Deserializer<T> = kafkaDeserializer()
// }
//
//}