Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Extract RPC code to separate package
- Loading branch information
1 parent
c5cf147
commit 3c38bc4
Showing
11 changed files
with
223 additions
and
194 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
171 changes: 171 additions & 0 deletions
171
latte/src/main/java/gg/beemo/latte/broker/Subclients.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,171 @@ | ||
| package gg.beemo.latte.broker | ||
|
|
||
| import com.squareup.moshi.JsonAdapter | ||
| import com.squareup.moshi.Moshi | ||
| import gg.beemo.latte.broker.rpc.RpcMessageHeaders | ||
| import gg.beemo.latte.logging.Log | ||
| import gg.beemo.latte.util.MoshiInstantAdapter | ||
| import gg.beemo.latte.util.MoshiJsLongAdapter | ||
| import gg.beemo.latte.util.MoshiUnitAdapter | ||
| import kotlinx.coroutines.CoroutineScope | ||
| import kotlinx.coroutines.coroutineScope | ||
| import java.time.Instant | ||
|
|
||
| data class BrokerClientOptions( | ||
| val useSafeJsLongs: Boolean = false, | ||
| ) | ||
|
|
||
| abstract class BaseSubclient( | ||
| protected val connection: BrokerConnection, | ||
| protected val client: BrokerClient, | ||
| val topic: String, | ||
| val key: String, | ||
| protected val options: BrokerClientOptions, | ||
| ) { | ||
|
|
||
| internal abstract fun destroy() | ||
|
|
||
| protected fun <T> createMoshiAdapter(type: Class<T>): JsonAdapter<T?> { | ||
| val mochi = if (options.useSafeJsLongs) safeJsMoshi else baseMoshi | ||
| return mochi.adapter(type).nullSafe() | ||
| } | ||
|
|
||
| companion object { | ||
| private val baseMoshi: Moshi = Moshi.Builder() | ||
| .add(Unit::class.java, MoshiUnitAdapter()) | ||
| .add(Instant::class.java, MoshiInstantAdapter()) | ||
| .build() | ||
| private val safeJsMoshi: Moshi = baseMoshi | ||
| .newBuilder() | ||
| .add(Long::class.java, MoshiJsLongAdapter()) | ||
| .build() | ||
| } | ||
|
|
||
| } | ||
|
|
||
| class ProducerSubclient<T>( | ||
| connection: BrokerConnection, | ||
| client: BrokerClient, | ||
| topic: String, | ||
| key: String, | ||
| options: BrokerClientOptions, | ||
| requestType: Class<T>, | ||
| private val isNullable: Boolean, | ||
| ) : BaseSubclient( | ||
| connection, | ||
| client, | ||
| topic, | ||
| key, | ||
| options, | ||
| ) { | ||
|
|
||
| private val log by Log | ||
| private val adapter: JsonAdapter<T?> = createMoshiAdapter(requestType) | ||
|
|
||
| override fun destroy() { | ||
| client.deregisterProducer(this) | ||
| } | ||
|
|
||
| suspend fun send( | ||
| data: T, | ||
| services: Set<String> = emptySet(), | ||
| instances: Set<String> = emptySet(), | ||
| ): MessageId { | ||
| val msg = BrokerMessage( | ||
| topic, | ||
| key, | ||
| data, | ||
| BrokerMessageHeaders( | ||
| connection, | ||
| targetServices = services, | ||
| targetInstances = instances, | ||
| ), | ||
| ) | ||
| @Suppress("UNCHECKED_CAST") | ||
| return internalSend(msg as AbstractBrokerMessage<T?>) | ||
| } | ||
|
|
||
| internal suspend fun internalSend(msg: AbstractBrokerMessage<T?>, bypassNullCheck: Boolean = false): MessageId { | ||
| if (!bypassNullCheck && !isNullable) { | ||
| requireNotNull(msg.value) { | ||
| "Cannot send null message for non-nullable type with key '$key' in topic '$topic'" | ||
| } | ||
| } | ||
| val strigifiedData = stringifyOutgoing(msg.value) | ||
| log.trace( | ||
| "Sending message {} with key '{}' in topic '{}' with value: {}", | ||
| msg.messageId, | ||
| key, | ||
| topic, | ||
| strigifiedData, | ||
| ) | ||
| return connection.send(topic, key, strigifiedData, msg.headers) | ||
| } | ||
|
|
||
| private fun stringifyOutgoing(data: T?): String { | ||
| return adapter.toJson(data) | ||
| } | ||
|
|
||
| } | ||
|
|
||
| class ConsumerSubclient<T>( | ||
| connection: BrokerConnection, | ||
| client: BrokerClient, | ||
| topic: String, | ||
| key: String, | ||
| options: BrokerClientOptions, | ||
| incomingType: Class<T>, | ||
| private val isNullable: Boolean, | ||
| private val callback: suspend CoroutineScope.(BaseBrokerMessage<T>) -> Unit, | ||
| ) : BaseSubclient( | ||
| connection, | ||
| client, | ||
| topic, | ||
| key, | ||
| options, | ||
| ) { | ||
|
|
||
| private val log by Log | ||
| private val adapter: JsonAdapter<T?> = createMoshiAdapter(incomingType) | ||
|
|
||
| override fun destroy() { | ||
| client.deregisterConsumer(this) | ||
| } | ||
|
|
||
| internal suspend fun onIncomingMessage( | ||
| value: String, | ||
| headers: BrokerMessageHeaders, | ||
| ) = coroutineScope { | ||
| val data = parseIncoming(value) | ||
| // Disable nullability enforcement for RPC exceptions. The caller has to deal with the unsafe typing now. | ||
| if (!isNullable && (headers !is RpcMessageHeaders || !headers.isException)) { | ||
| checkNotNull(data) { | ||
| "Received null message for non-nullable type with key '$key' in topic '$topic'" | ||
| } | ||
| } | ||
| val message = BrokerMessage(topic, key, data, headers) | ||
| log.trace( | ||
| "Received message {} with key '{}' in topic '{}' with value: {}", | ||
| headers.messageId, | ||
| key, | ||
| topic, | ||
| value, | ||
| ) | ||
| @Suppress("UNCHECKED_CAST") // Safe due to above null validation | ||
| val brokerMessage = message as BaseBrokerMessage<T> | ||
| try { | ||
| callback(brokerMessage) | ||
| } catch (ex: Exception) { | ||
| log.error( | ||
| "Uncaught consumer callback error while processing message ${headers.messageId} " + | ||
| "with key '$key' in topic '$topic'", | ||
| ex, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| private fun parseIncoming(json: String): T? { | ||
| return adapter.fromJson(json) | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.