forked from spring-projects/spring-kafka
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
spring-projectsGH-1189: support kotlin suspend
* Support kotlin suspend * Add kotlin suspend test, async request/reply with `@KafkaListener` `@KafkaHandler` and `@SendTo` * Fix async-returns.adoc and async warn log in `MessagingMessageListenerAdapter` * Add dependency `org.jetbrains.kotlinx:kotlinx-coroutines-reactor:1.7.3`
- Loading branch information
Zhiyang.Wang1
committed
Dec 21, 2023
1 parent
fcbc7d5
commit 3ea5faf
Showing
9 changed files
with
433 additions
and
12 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
50 changes: 50 additions & 0 deletions
50
.../java/org/springframework/kafka/annotation/ContinuationHandlerMethodArgumentResolver.java
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,50 @@ | ||
| /* | ||
| * Copyright 2023 the original author or authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.springframework.kafka.annotation; | ||
|
|
||
| /** | ||
| * No-op resolver for method arguments of type {@link kotlin.coroutines.Continuation}. | ||
| * <p> | ||
| * This class is similar to | ||
| * {@link org.springframework.messaging.handler.annotation.reactive.ContinuationHandlerMethodArgumentResolver} | ||
| * but for regular {@link HandlerMethodArgumentResolver} contract. | ||
| * | ||
| * @author Wang Zhiyang | ||
| * | ||
| * @since 3.1 | ||
| * | ||
| * @see org.springframework.messaging.handler.annotation.reactive.ContinuationHandlerMethodArgumentResolver | ||
| */ | ||
| import org.springframework.core.MethodParameter; | ||
| import org.springframework.messaging.Message; | ||
| import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; | ||
|
|
||
| import reactor.core.publisher.Mono; | ||
|
|
||
| public class ContinuationHandlerMethodArgumentResolver implements HandlerMethodArgumentResolver { | ||
|
|
||
| @Override | ||
| public boolean supportsParameter(MethodParameter parameter) { | ||
| return "kotlin.coroutines.Continuation".equals(parameter.getParameterType().getName()); | ||
| } | ||
|
|
||
| @Override | ||
| public Object resolveArgument(MethodParameter parameter, Message<?> message) { | ||
| return Mono.empty(); | ||
| } | ||
|
|
||
| } |
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
78 changes: 78 additions & 0 deletions
78
.../src/main/java/org/springframework/kafka/annotation/KafkaMessageHandlerMethodFactory.java
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,78 @@ | ||
| /* | ||
| * Copyright 2023 the original author or authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.springframework.kafka.annotation; | ||
|
|
||
| import java.lang.reflect.Method; | ||
| import java.util.List; | ||
|
|
||
| import org.springframework.core.KotlinDetector; | ||
| import org.springframework.messaging.converter.MessageConverter; | ||
| import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; | ||
| import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; | ||
| import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite; | ||
| import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; | ||
| import org.springframework.validation.Validator; | ||
|
|
||
| /** | ||
| * Extension of the {@link DefaultMessageHandlerMethodFactory} for Spring Kafka requirements. | ||
| * | ||
| * @author Wang Zhiyang | ||
| * | ||
| * @since 3.1 | ||
| */ | ||
| public class KafkaMessageHandlerMethodFactory extends DefaultMessageHandlerMethodFactory { | ||
|
|
||
| private final HandlerMethodArgumentResolverComposite argumentResolvers = | ||
| new HandlerMethodArgumentResolverComposite(); | ||
|
|
||
| private MessageConverter messageConverter; | ||
|
|
||
| private Validator validator; | ||
|
|
||
| @Override | ||
| public void setMessageConverter(MessageConverter messageConverter) { | ||
| super.setMessageConverter(messageConverter); | ||
| this.messageConverter = messageConverter; | ||
| } | ||
|
|
||
| @Override | ||
| public void setValidator(Validator validator) { | ||
| super.setValidator(validator); | ||
| this.validator = validator; | ||
| } | ||
|
|
||
| @Override | ||
| protected List<HandlerMethodArgumentResolver> initArgumentResolvers() { | ||
| List<HandlerMethodArgumentResolver> resolvers = super.initArgumentResolvers(); | ||
| if (KotlinDetector.isKotlinPresent()) { | ||
| // Insert before PayloadMethodArgumentResolver | ||
| resolvers.add(resolvers.size() - 1, new ContinuationHandlerMethodArgumentResolver()); | ||
| } | ||
| // Has to be at the end - look at PayloadMethodArgumentResolver documentation | ||
| resolvers.add(resolvers.size() - 1, new KafkaNullAwarePayloadArgumentResolver(this.messageConverter, this.validator)); | ||
| this.argumentResolvers.addResolvers(resolvers); | ||
| return resolvers; | ||
| } | ||
|
|
||
| @Override | ||
| public InvocableHandlerMethod createInvocableHandlerMethod(Object bean, Method method) { | ||
| InvocableHandlerMethod handlerMethod = new KotlinAwareInvocableHandlerMethod(bean, method); | ||
| handlerMethod.setMessageMethodArgumentResolvers(this.argumentResolvers); | ||
| return handlerMethod; | ||
| } | ||
|
|
||
| } |
49 changes: 49 additions & 0 deletions
49
...src/main/java/org/springframework/kafka/annotation/KotlinAwareInvocableHandlerMethod.java
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,49 @@ | ||
| /* | ||
| * Copyright 2023 the original author or authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.springframework.kafka.annotation; | ||
|
|
||
| import java.lang.reflect.Method; | ||
|
|
||
| import org.springframework.core.CoroutinesUtils; | ||
| import org.springframework.core.KotlinDetector; | ||
| import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; | ||
|
|
||
| /** | ||
| * An {@link InvocableHandlerMethod} extension for supporting Kotlin {@code suspend} function. | ||
| * | ||
| * @author Wang Zhiyang | ||
| * | ||
| * @since 3.1 | ||
| */ | ||
| public class KotlinAwareInvocableHandlerMethod extends InvocableHandlerMethod { | ||
|
|
||
| public KotlinAwareInvocableHandlerMethod(Object bean, Method method) { | ||
| super(bean, method); | ||
| } | ||
|
|
||
| @Override | ||
| protected Object doInvoke(Object... args) throws Exception { | ||
| Method method = getBridgedMethod(); | ||
| if (KotlinDetector.isSuspendingFunction(method)) { | ||
| return CoroutinesUtils.invokeSuspendingFunction(method, getBean(), args); | ||
| } | ||
| else { | ||
| return super.doInvoke(args); | ||
| } | ||
| } | ||
|
|
||
| } |
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
Oops, something went wrong.