mirror of
https://github.com/InsanusMokrassar/TelegramBotAPI.git
synced 2025-09-08 09:42:59 +00:00
migration
This commit is contained in:
@@ -0,0 +1,60 @@
|
||||
package com.github.insanusmokrassar.TelegramBotAPI.updateshandlers
|
||||
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.MediaGroupUpdates.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.utils.extensions.UpdateReceiver
|
||||
import kotlinx.coroutines.channels.BroadcastChannel
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.asFlow
|
||||
|
||||
private fun <T> BroadcastChannel<T>.createUpdateReceiver(): UpdateReceiver<T> = ::send
|
||||
|
||||
class FlowsUpdatesFilter {
|
||||
private val messageChannel: BroadcastChannel<MessageUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val messageMediaGroupChannel: BroadcastChannel<MessageMediaGroupUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val editedMessageChannel: BroadcastChannel<EditMessageUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val editedMessageMediaGroupChannel: BroadcastChannel<EditMessageMediaGroupUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val channelPostChannel: BroadcastChannel<ChannelPostUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val channelPostMediaGroupChannel: BroadcastChannel<ChannelPostMediaGroupUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val editedChannelPostChannel: BroadcastChannel<EditChannelPostUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val editedChannelPostMediaGroupChannel: BroadcastChannel<EditChannelPostMediaGroupUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val chosenInlineResultChannel: BroadcastChannel<ChosenInlineResultUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val inlineQueryChannel: BroadcastChannel<InlineQueryUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val callbackQueryChannel: BroadcastChannel<CallbackQueryUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val shippingQueryChannel: BroadcastChannel<ShippingQueryUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val preCheckoutQueryChannel: BroadcastChannel<PreCheckoutQueryUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
private val pollChannel: BroadcastChannel<PollUpdate> = BroadcastChannel(Channel.CONFLATED)
|
||||
|
||||
val filter = UpdatesFilter(
|
||||
messageChannel.createUpdateReceiver(),
|
||||
messageMediaGroupChannel.createUpdateReceiver(),
|
||||
editedMessageChannel.createUpdateReceiver(),
|
||||
editedMessageMediaGroupChannel.createUpdateReceiver(),
|
||||
channelPostChannel.createUpdateReceiver(),
|
||||
channelPostMediaGroupChannel.createUpdateReceiver(),
|
||||
editedChannelPostChannel.createUpdateReceiver(),
|
||||
editedChannelPostMediaGroupChannel.createUpdateReceiver(),
|
||||
chosenInlineResultChannel.createUpdateReceiver(),
|
||||
inlineQueryChannel.createUpdateReceiver(),
|
||||
callbackQueryChannel.createUpdateReceiver(),
|
||||
shippingQueryChannel.createUpdateReceiver(),
|
||||
preCheckoutQueryChannel.createUpdateReceiver(),
|
||||
pollChannel.createUpdateReceiver()
|
||||
)
|
||||
|
||||
val messageFlow: Flow<MessageUpdate> = messageChannel.asFlow()
|
||||
val messageMediaGroupFlow: Flow<MessageMediaGroupUpdate> = messageMediaGroupChannel.asFlow()
|
||||
val editedMessageFlow: Flow<EditMessageUpdate> = editedMessageChannel.asFlow()
|
||||
val editedMessageMediaGroupFlow: Flow<EditMessageMediaGroupUpdate> = editedMessageMediaGroupChannel.asFlow()
|
||||
val channelPostFlow: Flow<ChannelPostUpdate> = channelPostChannel.asFlow()
|
||||
val channelPostMediaGroupFlow: Flow<ChannelPostMediaGroupUpdate> = channelPostMediaGroupChannel.asFlow()
|
||||
val editedChannelPostFlow: Flow<EditChannelPostUpdate> = editedChannelPostChannel.asFlow()
|
||||
val editedChannelPostMediaGroupFlow: Flow<EditChannelPostMediaGroupUpdate> = editedChannelPostMediaGroupChannel.asFlow()
|
||||
val chosenInlineResultFlow: Flow<ChosenInlineResultUpdate> = chosenInlineResultChannel.asFlow()
|
||||
val inlineQueryFlow: Flow<InlineQueryUpdate> = inlineQueryChannel.asFlow()
|
||||
val callbackQueryFlow: Flow<CallbackQueryUpdate> = callbackQueryChannel.asFlow()
|
||||
val shippingQueryFlow: Flow<ShippingQueryUpdate> = shippingQueryChannel.asFlow()
|
||||
val preCheckoutQueryFlow: Flow<PreCheckoutQueryUpdate> = preCheckoutQueryChannel.asFlow()
|
||||
val pollFlow: Flow<PollUpdate> = pollChannel.asFlow()
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package com.github.insanusmokrassar.TelegramBotAPI.updateshandlers
|
||||
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.bot.Ktor.KtorRequestsExecutor
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.bot.RequestsExecutor
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.bot.UpdatesPoller
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.requests.GetUpdates
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.requests.webhook.DeleteWebhook
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.ALL_UPDATES_LIST
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.UpdateIdentifier
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.message.abstracts.MediaGroupMessage
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.abstracts.BaseMessageUpdate
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.abstracts.Update
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.utils.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.utils.extensions.UpdateReceiver
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.utils.extensions.executeUnsafe
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.HttpClientEngine
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
|
||||
fun KtorUpdatesPoller(
|
||||
telegramAPIUrlsKeeper: TelegramAPIUrlsKeeper,
|
||||
engine: HttpClientEngine,
|
||||
timeoutSeconds: Int? = null,
|
||||
oneTimeUpdatesLimit: Int? = null,
|
||||
allowedUpdates: List<String> = ALL_UPDATES_LIST,
|
||||
exceptionsHandler: (Exception) -> Boolean = { true },
|
||||
updatesReceiver: UpdateReceiver<Update>
|
||||
): KtorUpdatesPoller {
|
||||
val executor = KtorRequestsExecutor(
|
||||
telegramAPIUrlsKeeper,
|
||||
HttpClient(engine)
|
||||
)
|
||||
|
||||
return KtorUpdatesPoller(
|
||||
executor,
|
||||
timeoutSeconds,
|
||||
oneTimeUpdatesLimit,
|
||||
allowedUpdates,
|
||||
exceptionsHandler,
|
||||
updatesReceiver
|
||||
)
|
||||
}
|
||||
|
||||
class KtorUpdatesPoller(
|
||||
private val executor: RequestsExecutor,
|
||||
private val timeoutSeconds: Int? = null,
|
||||
private val oneTimeUpdatesLimit: Int? = null,
|
||||
private val allowedUpdates: List<String> = ALL_UPDATES_LIST,
|
||||
private val exceptionsHandler: (Exception) -> Boolean = { true },
|
||||
private val updatesReceiver: UpdateReceiver<Update>
|
||||
) : UpdatesPoller {
|
||||
private var lastHandledUpdate: UpdateIdentifier = 0L
|
||||
private val mediaGroup: MutableList<BaseMessageUpdate> = mutableListOf()
|
||||
|
||||
private var pollerJob: Job? = null
|
||||
|
||||
private suspend fun sendToBlock(data: Update) {
|
||||
updatesReceiver(data)
|
||||
lastHandledUpdate = data.updateId
|
||||
}
|
||||
|
||||
private suspend fun pushMediaGroupUpdate(update: BaseMessageUpdate? = null) {
|
||||
val inputMediaGroupId = (update ?.data as? MediaGroupMessage) ?.mediaGroupId
|
||||
if (mediaGroup.isNotEmpty() && inputMediaGroupId ?.equals(mediaGroup.mediaGroupId) != true) {
|
||||
mediaGroup.sortBy { it.updateId }
|
||||
mediaGroup.convertWithMediaGroupUpdates().forEach {
|
||||
sendToBlock(it)
|
||||
}
|
||||
mediaGroup.clear()
|
||||
}
|
||||
inputMediaGroupId ?.let {
|
||||
mediaGroup.add(update)
|
||||
} ?: sendToBlock(update ?: return)
|
||||
}
|
||||
|
||||
private suspend fun getUpdates(): List<Update> {
|
||||
return executor.execute(
|
||||
GetUpdates(
|
||||
lastHandledUpdate + 1, // incremented because offset counted from 1 when updates id from 0
|
||||
oneTimeUpdatesLimit,
|
||||
timeoutSeconds,
|
||||
allowedUpdates
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun handleUpdates(updates: List<Update>) {
|
||||
for (update in updates) {
|
||||
(update as? BaseMessageUpdate) ?.let {
|
||||
if (it.data is MediaGroupMessage) {
|
||||
pushMediaGroupUpdate(it)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
} ?:let {
|
||||
pushMediaGroupUpdate()
|
||||
sendToBlock(update)
|
||||
}
|
||||
}
|
||||
|
||||
pushMediaGroupUpdate()
|
||||
}
|
||||
|
||||
private val startStopScope = CoroutineScope(Dispatchers.Default)
|
||||
private val startStopQueue = Channel<CoroutineScope?>(2)
|
||||
private val startStopJob = startStopScope.launch {
|
||||
for (scope in startStopQueue) {
|
||||
scope ?.also {
|
||||
pollerJob ?: scope.launch {
|
||||
executor.executeUnsafe(DeleteWebhook())
|
||||
while (isActive) {
|
||||
try {
|
||||
val updates = getUpdates()
|
||||
handleUpdates(updates)
|
||||
} catch (e: Exception) {
|
||||
if (exceptionsHandler(e)) {
|
||||
continue
|
||||
} else {
|
||||
close()
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}.also {
|
||||
pollerJob = it
|
||||
}
|
||||
} ?: also {
|
||||
startStopQueue.close()
|
||||
pollerJob ?.cancel()
|
||||
startStopScope.cancel()
|
||||
return@launch
|
||||
}
|
||||
}
|
||||
}
|
||||
override fun start(scope: CoroutineScope) {
|
||||
startStopQueue.offer(scope)
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
startStopQueue.offer(null)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package com.github.insanusmokrassar.TelegramBotAPI.updateshandlers
|
||||
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.MediaGroupUpdates.*
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.types.update.abstracts.Update
|
||||
import com.github.insanusmokrassar.TelegramBotAPI.utils.extensions.UpdateReceiver
|
||||
|
||||
data class UpdatesFilter(
|
||||
private val messageCallback: UpdateReceiver<MessageUpdate>? = null,
|
||||
private val messageMediaGroupCallback: UpdateReceiver<MessageMediaGroupUpdate>? = null,
|
||||
private val editedMessageCallback: UpdateReceiver<EditMessageUpdate>? = null,
|
||||
private val editedMessageMediaGroupCallback: UpdateReceiver<EditMessageMediaGroupUpdate>? = null,
|
||||
private val channelPostCallback: UpdateReceiver<ChannelPostUpdate>? = null,
|
||||
private val channelPostMediaGroupCallback: UpdateReceiver<ChannelPostMediaGroupUpdate>? = null,
|
||||
private val editedChannelPostCallback: UpdateReceiver<EditChannelPostUpdate>? = null,
|
||||
private val editedChannelPostMediaGroupCallback: UpdateReceiver<EditChannelPostMediaGroupUpdate>? = null,
|
||||
private val chosenInlineResultCallback: UpdateReceiver<ChosenInlineResultUpdate>? = null,
|
||||
private val inlineQueryCallback: UpdateReceiver<InlineQueryUpdate>? = null,
|
||||
private val callbackQueryCallback: UpdateReceiver<CallbackQueryUpdate>? = null,
|
||||
private val shippingQueryCallback: UpdateReceiver<ShippingQueryUpdate>? = null,
|
||||
private val preCheckoutQueryCallback: UpdateReceiver<PreCheckoutQueryUpdate>? = null,
|
||||
private val pollUpdateCallback: UpdateReceiver<PollUpdate>? = null
|
||||
) {
|
||||
val asUpdateReceiver: UpdateReceiver<Update> = this::invoke
|
||||
val allowedUpdates = listOfNotNull(
|
||||
(messageCallback ?: messageMediaGroupCallback) ?.let { UPDATE_MESSAGE },
|
||||
(editedMessageCallback ?: editedMessageMediaGroupCallback) ?.let { UPDATE_EDITED_MESSAGE },
|
||||
(channelPostCallback ?: channelPostMediaGroupCallback) ?.let { UPDATE_CHANNEL_POST },
|
||||
(editedChannelPostCallback ?: editedChannelPostMediaGroupCallback) ?.let { UPDATE_EDITED_CHANNEL_POST },
|
||||
chosenInlineResultCallback ?.let { UPDATE_CHOSEN_INLINE_RESULT },
|
||||
inlineQueryCallback ?.let { UPDATE_INLINE_QUERY },
|
||||
callbackQueryCallback ?.let { UPDATE_CALLBACK_QUERY },
|
||||
shippingQueryCallback ?.let { UPDATE_SHIPPING_QUERY },
|
||||
preCheckoutQueryCallback ?.let { UPDATE_PRE_CHECKOUT_QUERY },
|
||||
pollUpdateCallback ?.let { UPDATE_POLL }
|
||||
)
|
||||
|
||||
suspend fun invoke(update: Update) {
|
||||
when (update) {
|
||||
is MessageUpdate -> messageCallback ?.invoke(update)
|
||||
is MessageMediaGroupUpdate -> messageMediaGroupCallback ?.also { receiver ->
|
||||
receiver(update)
|
||||
} ?: messageCallback ?.also { receiver ->
|
||||
update.origins.mapNotNull { it as? MessageUpdate }.forEach {
|
||||
receiver(it)
|
||||
}
|
||||
}
|
||||
is EditMessageUpdate -> editedMessageCallback ?.invoke(update)
|
||||
is EditMessageMediaGroupUpdate -> editedMessageMediaGroupCallback ?.also { receiver ->
|
||||
receiver(update)
|
||||
} ?: editedMessageCallback ?.also { receiver ->
|
||||
receiver(update.origin)
|
||||
}
|
||||
is ChannelPostUpdate -> channelPostCallback ?.invoke(update)
|
||||
is ChannelPostMediaGroupUpdate -> channelPostMediaGroupCallback ?.also { receiver ->
|
||||
receiver(update)
|
||||
} ?: channelPostCallback ?.also { receiver ->
|
||||
update.origins.mapNotNull { it as? ChannelPostUpdate }.forEach {
|
||||
receiver(it)
|
||||
}
|
||||
}
|
||||
is EditChannelPostUpdate -> editedChannelPostCallback ?.invoke(update)
|
||||
is EditChannelPostMediaGroupUpdate -> editedChannelPostMediaGroupCallback ?.also { receiver ->
|
||||
receiver(update)
|
||||
} ?: editedChannelPostCallback ?.also { receiver ->
|
||||
receiver(update.origin)
|
||||
}
|
||||
is ChosenInlineResultUpdate -> chosenInlineResultCallback ?.invoke(update)
|
||||
is InlineQueryUpdate -> inlineQueryCallback ?.invoke(update)
|
||||
is CallbackQueryUpdate -> callbackQueryCallback ?.invoke(update)
|
||||
is ShippingQueryUpdate -> shippingQueryCallback ?.invoke(update)
|
||||
is PreCheckoutQueryUpdate -> preCheckoutQueryCallback ?.invoke(update)
|
||||
is PollUpdate -> pollUpdateCallback ?.invoke(update)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun createSimpleUpdateFilter(
|
||||
messageCallback: UpdateReceiver<MessageUpdate>? = null,
|
||||
mediaGroupCallback: UpdateReceiver<MediaGroupUpdate>? = null,
|
||||
editedMessageCallback: UpdateReceiver<EditMessageUpdate>? = null,
|
||||
channelPostCallback: UpdateReceiver<ChannelPostUpdate>? = null,
|
||||
editedChannelPostCallback: UpdateReceiver<EditChannelPostUpdate>? = null,
|
||||
chosenInlineResultCallback: UpdateReceiver<ChosenInlineResultUpdate>? = null,
|
||||
inlineQueryCallback: UpdateReceiver<InlineQueryUpdate>? = null,
|
||||
callbackQueryCallback: UpdateReceiver<CallbackQueryUpdate>? = null,
|
||||
shippingQueryCallback: UpdateReceiver<ShippingQueryUpdate>? = null,
|
||||
preCheckoutQueryCallback: UpdateReceiver<PreCheckoutQueryUpdate>? = null,
|
||||
pollCallback: UpdateReceiver<PollUpdate>? = null
|
||||
): UpdatesFilter = UpdatesFilter(
|
||||
messageCallback = messageCallback,
|
||||
messageMediaGroupCallback = mediaGroupCallback,
|
||||
editedMessageCallback = editedMessageCallback,
|
||||
editedMessageMediaGroupCallback = mediaGroupCallback,
|
||||
channelPostCallback = channelPostCallback,
|
||||
channelPostMediaGroupCallback = mediaGroupCallback,
|
||||
editedChannelPostCallback = editedChannelPostCallback,
|
||||
editedChannelPostMediaGroupCallback = mediaGroupCallback,
|
||||
chosenInlineResultCallback = chosenInlineResultCallback,
|
||||
inlineQueryCallback = inlineQueryCallback,
|
||||
callbackQueryCallback = callbackQueryCallback,
|
||||
shippingQueryCallback = shippingQueryCallback,
|
||||
preCheckoutQueryCallback = preCheckoutQueryCallback,
|
||||
pollUpdateCallback = pollCallback
|
||||
)
|
||||
Reference in New Issue
Block a user