Kotlin 支持
该框架也得到了改进,以支持函数的 Kotlin lambda,因此现在您可以结合使用 Kotlin 语言和 Spring 集成流定义:
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { print(it) }
}
@Bean
@InboundChannelAdapter(value = "counterChannel",
poller = Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
fun kotlinSupplier(): () -> String {
return { "baz" }
}
Kotlin 协程
从版本 6.0 开始, Spring 集成提供了对 Kotlin 协程的支持。
现在suspend
functions 和kotlinx.coroutines.Deferred
& kotlinx.coroutines.flow.Flow
返回类型可用于服务方法:
@ServiceActivator(inputChannel = "suspendServiceChannel", outputChannel = "resultChannel")
suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
@ServiceActivator(inputChannel = "flowServiceChannel", outputChannel = "resultChannel", async = "true")
fun flowServiceFunction(payload: String) =
flow {
for (i in 1..3) {
emit("$payload #$i")
}
}
该框架将它们视为 Reactive Streams 交互,并使用ReactiveAdapterRegistry
转换为相应的Mono
和Flux
反应器类型。
这样的函数 reply 将在 reply channel 中处理,如果它是一个ReactiveStreamsSubscribableChannel
或作为CompletableFuture
在相应的回调中。
具有Flow result 不是async 默认情况下,在@ServiceActivator 所以Flow 实例作为回复消息有效负载生成。
目标应用程序负责将此对象作为协程处理或将其转换为Flux 分别。 |
这@MessagingGateway
interface 方法也可以用suspend
修饰符。
该框架利用Mono
internally 使用下游流执行请求-回复。
这样的Mono
result 由MonoKt.awaitSingleOrNull()
API 来实现kotlin.coroutines.Continuation
argument fo the calledsuspend
网关的功能:
@MessagingGateway(defaultRequestChannel = "suspendRequestChannel")
interface SuspendFunGateway {
suspend fun suspendGateway(payload: String): String
}
根据 Kotlin 语言要求,此方法必须作为协程调用:
@Autowired
private lateinit var suspendFunGateway: SuspendFunGateway
fun someServiceMethod() {
runBlocking {
val reply = suspendFunGateway.suspendGateway("test suspend gateway")
}
}