-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: integration flow is sync for messages from kafka (#127)
* refactor: created sync gateway * refactor: added send message exception * refactor: moved send messages form kafka to sync integration flows * refactor: added exceptions throwing for telegram api errors * refactor: controller use sync integration flow * chore: updated version to 1.3.6
- Loading branch information
Showing
8 changed files
with
156 additions
and
78 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
120 changes: 120 additions & 0 deletions
120
telegram_bot/src/main/java/ru/dankoy/telegrambot/config/integration/FlowSyncConfig.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,120 @@ | ||
| package ru.dankoy.telegrambot.config.integration; | ||
|
|
||
| import io.github.resilience4j.ratelimiter.RateLimiterConfig; | ||
| import java.time.Duration; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
| import org.springframework.integration.channel.DirectChannel; | ||
| import org.springframework.integration.dsl.IntegrationFlow; | ||
| import org.springframework.integration.handler.advice.RateLimiterRequestHandlerAdvice; | ||
| import org.springframework.messaging.MessageChannel; | ||
| import ru.dankoy.telegrambot.core.domain.message.ChannelSubscriptionMessage; | ||
| import ru.dankoy.telegrambot.core.domain.message.CommunitySubscriptionMessage; | ||
| import ru.dankoy.telegrambot.core.domain.message.CoubMessage; | ||
| import ru.dankoy.telegrambot.core.domain.message.TagSubscriptionMessage; | ||
| import ru.dankoy.telegrambot.core.service.bot.TelegramBot; | ||
| import ru.dankoy.telegrambot.core.service.reply.ReplyCreatorService; | ||
|
|
||
| /** | ||
| * @author dankoy | ||
| * <p>Should work in sync mode. It was made to fix bug <a | ||
| * href="https://github.com/Dankoy/jforwarder/issues/126">...</a> | ||
| */ | ||
| @Slf4j | ||
| @Configuration | ||
| public class FlowSyncConfig { | ||
|
|
||
| @Bean | ||
| public MessageChannel subscriptionMessagesChannel() { | ||
| return new DirectChannel(); | ||
| } | ||
|
|
||
| @Bean | ||
| public MessageChannel sendMessageDirectChannel() { | ||
| // this channel send messages as soon they created. | ||
| // With queue integration flow waited 5 seconds every time before handle message. | ||
| return new DirectChannel(); | ||
| } | ||
|
|
||
| @Bean | ||
| public MessageChannel communitySubscriptionSendDirectChannel() { | ||
| return new DirectChannel(); | ||
| } | ||
|
|
||
| @Bean | ||
| public MessageChannel tagSubscriptionSendDirectChannel() { | ||
| return new DirectChannel(); | ||
| } | ||
|
|
||
| @Bean | ||
| public MessageChannel channelSubscriptionSendDirectChannel() { | ||
| return new DirectChannel(); | ||
| } | ||
|
|
||
| // rate limiter for one message in 2 second | ||
| @Bean | ||
| public RateLimiterRequestHandlerAdvice rateLimiterRequestOneInTwoSecondsHandlerAdvice() { | ||
| return new RateLimiterRequestHandlerAdvice( | ||
| RateLimiterConfig.custom() | ||
| .limitRefreshPeriod(Duration.ofSeconds(2)) | ||
| .limitForPeriod(1) | ||
| .build()); | ||
| } | ||
|
|
||
| @Bean | ||
| public IntegrationFlow sendMessageSyncFlow(TelegramBot telegramBot) { | ||
| return IntegrationFlow.from(sendMessageDirectChannel()) | ||
| .handle( | ||
| telegramBot, | ||
| "sendMessage", | ||
| c -> c.advice(rateLimiterRequestOneInTwoSecondsHandlerAdvice())) | ||
| .get(); | ||
| } | ||
|
|
||
| @Bean | ||
| public IntegrationFlow subscriptionMessagesFlow() { | ||
|
|
||
| return IntegrationFlow.from(subscriptionMessagesChannel()) | ||
| .<CoubMessage, Class<?>>route( | ||
| CoubMessage::getClass, | ||
| m -> | ||
| m.channelMapping( | ||
| CommunitySubscriptionMessage.class, | ||
| "communitySubscriptionSendDirectChannel") | ||
| .channelMapping( | ||
| TagSubscriptionMessage.class, "tagSubscriptionSendDirectChannel") | ||
| .channelMapping( | ||
| ChannelSubscriptionMessage.class, "channelSubscriptionSendDirectChannel")) | ||
| .get(); | ||
| } | ||
|
|
||
| @Bean | ||
| public IntegrationFlow communitySubscriptionSendMessageFlow( | ||
| ReplyCreatorService replyCreatorService) { | ||
|
|
||
| return IntegrationFlow.from(communitySubscriptionSendDirectChannel()) | ||
| .handle(replyCreatorService, "createCommunitySubscriptionMessage") | ||
| .channel(sendMessageDirectChannel()) | ||
| .get(); | ||
| } | ||
|
|
||
| @Bean | ||
| public IntegrationFlow tagSubscriptionSendMessageFlow(ReplyCreatorService replyCreatorService) { | ||
|
|
||
| return IntegrationFlow.from(tagSubscriptionSendDirectChannel()) | ||
| .handle(replyCreatorService, "createTagSubscriptionMessage") | ||
| .channel(sendMessageDirectChannel()) | ||
| .get(); | ||
| } | ||
|
|
||
| @Bean | ||
| public IntegrationFlow channelSubscriptionSendMessageFlow( | ||
| ReplyCreatorService replyCreatorService) { | ||
|
|
||
| return IntegrationFlow.from(channelSubscriptionSendDirectChannel()) | ||
| .handle(replyCreatorService, "createChannelSubscriptionMessage") | ||
| .channel(sendMessageDirectChannel()) | ||
| .get(); | ||
| } | ||
| } |
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
12 changes: 12 additions & 0 deletions
12
...gram_bot/src/main/java/ru/dankoy/telegrambot/core/exceptions/BotSendMessageException.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,12 @@ | ||
| package ru.dankoy.telegrambot.core.exceptions; | ||
|
|
||
| public class BotSendMessageException extends BotException { | ||
|
|
||
| public BotSendMessageException(String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| public BotSendMessageException(String message, Exception e) { | ||
| super(message, e); | ||
| } | ||
| } |
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
15 changes: 15 additions & 0 deletions
15
telegram_bot/src/main/java/ru/dankoy/telegrambot/core/gateway/BotMessageSyncGateway.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,15 @@ | ||
| package ru.dankoy.telegrambot.core.gateway; | ||
|
|
||
| import org.springframework.integration.annotation.Gateway; | ||
| import org.springframework.integration.annotation.MessagingGateway; | ||
| import ru.dankoy.telegrambot.core.domain.message.CoubMessage; | ||
|
|
||
| // To make sync gateway all their channels must be sync (direct channel), not executors, also don't | ||
| // include errorChannel name here | ||
|
|
||
| @MessagingGateway | ||
| public interface BotMessageSyncGateway { | ||
|
|
||
| @Gateway(requestChannel = "subscriptionMessagesChannel") | ||
| void process(CoubMessage communitySubscriptionMessage); | ||
| } |
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