Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[WIP] Make header class more flexible and extendable
- Loading branch information
1 parent
e0d01b6
commit aad14e6
Showing
12 changed files
with
285 additions
and
108 deletions.
There are no files selected for viewing
37 changes: 0 additions & 37 deletions
37
latte/src/main/java/gg/beemo/latte/broker/BaseBrokerMessageHeaders.kt
This file was deleted.
Oops, something went wrong.
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
29 changes: 26 additions & 3 deletions
29
latte/src/main/java/gg/beemo/latte/broker/BrokerMessage.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 |
|---|---|---|
| @@ -1,14 +1,37 @@ | ||
| package gg.beemo.latte.broker | ||
|
|
||
| data class BrokerMessage<T>( | ||
| val client: BrokerClient, | ||
| open class BrokerMessage<T, H : BrokerMessageHeaders>( | ||
| val topic: String, | ||
| val key: String, | ||
| val value: T, | ||
| val headers: BaseBrokerMessageHeaders | ||
| val headers: H | ||
| ) { | ||
|
|
||
| val messageId: String | ||
| get() = headers.messageId | ||
|
|
||
| internal fun <ResponseT> toRpcRequestMessage( | ||
| updateSender: suspend (ResponseT, RpcStatus) -> Unit, | ||
| ): RpcRequestMessage<T, ResponseT, H> { | ||
| return RpcRequestMessage(topic, key, value, headers, updateSender) | ||
| } | ||
|
|
||
| } | ||
|
|
||
| typealias BaseBrokerMessage<T> = BrokerMessage<T, BrokerMessageHeaders> | ||
| typealias BaseRpcRequestMessage<RequestT, ResponseT> = RpcRequestMessage<RequestT, ResponseT, BrokerMessageHeaders> | ||
| typealias RpcResponseMessage<T> = BrokerMessage<T, RpcMessageHeaders> | ||
|
|
||
| class RpcRequestMessage<RequestT, ResponseT, H : BrokerMessageHeaders>( | ||
| topic: String, | ||
| key: String, | ||
| value: RequestT, | ||
| headers: H, | ||
| private val updateSender: suspend (ResponseT, RpcStatus) -> Unit, | ||
| ) : BrokerMessage<RequestT, H>(topic, key, value, headers) { | ||
|
|
||
| suspend fun sendUpdate(response: ResponseT, status: RpcStatus) { | ||
| updateSender(response, status) | ||
| } | ||
|
|
||
| } |
91 changes: 91 additions & 0 deletions
91
latte/src/main/java/gg/beemo/latte/broker/BrokerMessageHeaders.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,91 @@ | ||
| package gg.beemo.latte.broker | ||
|
|
||
| import java.util.UUID | ||
|
|
||
| open class BrokerMessageHeaders(val headers: Map<String, String>) { | ||
|
|
||
| val sourceService: String by lazy { | ||
| headers.getOrThrow(HEADER_SOURCE_SERVICE) | ||
| } | ||
| val sourceInstance: String by lazy { | ||
| headers.getOrThrow(HEADER_SOURCE_INSTANCE) | ||
| } | ||
| val targetServices: Set<String> by lazy { | ||
| splitToSet(headers.getOrDefault(HEADER_TARGET_SERVICES, "")) | ||
| } | ||
| val targetInstances: Set<String> by lazy { | ||
| splitToSet(headers.getOrDefault(HEADER_TARGET_INSTANCES, "")) | ||
| } | ||
| val messageId: MessageId by lazy { | ||
| headers.getOrThrow(HEADER_MESSAGE_ID) | ||
| } | ||
|
|
||
| constructor( | ||
| sourceService: String, | ||
| sourceInstance: String, | ||
| targetServices: Set<String>, | ||
| targetInstances: Set<String>, | ||
| ) : this( | ||
| createHeadersMap( | ||
| sourceService, | ||
| sourceInstance, | ||
| targetServices, | ||
| targetInstances, | ||
| null, | ||
| ) | ||
| ) | ||
|
|
||
| constructor( | ||
| connection: BrokerConnection, | ||
| targetServices: Set<String>, | ||
| targetInstances: Set<String>, | ||
| ) : this( | ||
| connection.serviceName, | ||
| connection.instanceId, | ||
| targetServices, | ||
| targetInstances, | ||
| ) | ||
|
|
||
| companion object { | ||
|
|
||
| private const val HEADER_SOURCE_SERVICE = "source-service" | ||
| private const val HEADER_SOURCE_INSTANCE = "source-instance" | ||
| private const val HEADER_TARGET_SERVICES = "target-services" | ||
| private const val HEADER_TARGET_INSTANCES = "target-instances" | ||
| private const val HEADER_MESSAGE_ID = "message-id" | ||
|
|
||
| @JvmStatic | ||
| protected fun createHeadersMap( | ||
| sourceService: String, | ||
| sourceInstance: String, | ||
| targetServices: Set<String>, | ||
| targetInstances: Set<String>, | ||
| messageId: MessageId?, | ||
| extra: Map<String, String> = emptyMap(), | ||
| ): Map<String, String> { | ||
| return mapOf( | ||
| HEADER_SOURCE_SERVICE to sourceService, | ||
| HEADER_SOURCE_INSTANCE to sourceInstance, | ||
| HEADER_TARGET_SERVICES to joinToString(targetServices), | ||
| HEADER_TARGET_INSTANCES to joinToString(targetInstances), | ||
| HEADER_MESSAGE_ID to (messageId ?: UUID.randomUUID().toString()), | ||
| ) | ||
| } | ||
|
|
||
| @JvmStatic | ||
| protected fun splitToSet(value: String): Set<String> { | ||
| return value.split(",").filter { it.isNotEmpty() }.toSet() | ||
| } | ||
|
|
||
| @JvmStatic | ||
| protected fun joinToString(value: Set<String>): String { | ||
| return value.joinToString(",") | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
|
|
||
| internal fun Map<String, String>.getOrThrow(key: String): String { | ||
| return get(key) ?: throw IllegalArgumentException("Missing broker message header '$key'") | ||
| } |
Oops, something went wrong.