KafkaMessageBroker

class KafkaMessageBroker<K : Any, V : Any>(consumerProps: Map<String, Any>? = null, producerProps: Map<String, Any>? = null, keyClass: Class<K>, valClass: Class<V>, needSubscribe: Boolean = true) : MessageBroker<K, V>

Constructors

constructor(consumerProps: Map<String, Any>? = null, producerProps: Map<String, Any>? = null, keyClass: Class<K>, valClass: Class<V>, needSubscribe: Boolean = true)

open override fun createTopic(topic: String, pCount: Int)

open override fun <KC : Any, VC : Any> publish(topic: String, brokerMsg: BrokerMsg<KC, VC>)

open override fun <KC : Any, VC : Any> publishAll(topic: String, records: List<BrokerMsg<KC, VC>>)

open override fun publishAllOld(topic: String, records: List<Pair<K?, V>>)

open override fun publishOld(topic: String, key: K?, message: V)

open override fun <KC : Any, VC : Any> subscribeAll(topic: String, keyClass: Class<KC>, valueClass: Class<VC>, onReceive: subscribeCallback<KC, VC>)

inline fun <KC : Any, VC : Any> MessageBroker<*, *>.subscribeAll(topic: String, noinline onReceive: subscribeCallback<KC, VC>)

open override fun subscribeAllOld(topic: String, onReceive: (t: String, List<Pair<K?, V>>) -> Unit)

open override fun <KC : Any, VC : Any> subscribeWithFilter(topic: String, keyClass: Class<KC>, valueClass: Class<VC>, filter: (KC, VC) -> Boolean, onReceive: subscribeCallback<KC, VC>)

inline fun <KC : Any, VC : Any> MessageBroker<*, *>.subscribeWithFilter(topic: String, noinline filter: (KC, VC) -> Boolean, noinline onReceive: subscribeCallback<KC, VC>)

open override fun subscribeWithFilterOld(topic: String, filter: (K?, V) -> Boolean, onReceive: (t: String, List<Pair<K?, V>>) -> Unit)

open override fun unsubscribe()