python-telegram-bot/telegram/ext/messagequeue.py
2022-01-03 08:15:18 +01:00

334 lines
14 KiB
Python

#!/usr/bin/env python
#
# Module author:
# Tymofii A. Khodniev (thodnev) <thodnev@mail.ru>
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2022
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/]
"""A throughput-limiting message processor for Telegram bots."""
import functools
import queue as q
import threading
import time
import warnings
from typing import TYPE_CHECKING, Callable, List, NoReturn
from telegram.ext.utils.promise import Promise
from telegram.utils.deprecate import TelegramDeprecationWarning
if TYPE_CHECKING:
from telegram import Bot
# We need to count < 1s intervals, so the most accurate timer is needed
curtime = time.perf_counter
class DelayQueueError(RuntimeError):
"""Indicates processing errors."""
__slots__ = ()
class DelayQueue(threading.Thread):
"""
Processes callbacks from queue with specified throughput limits. Creates a separate thread to
process callbacks with delays.
.. deprecated:: 13.3
:class:`telegram.ext.DelayQueue` in its current form is deprecated and will be reinvented
in a future release. See `this thread <https://git.io/JtDbF>`_ for a list of known bugs.
Args:
queue (:obj:`Queue`, optional): Used to pass callbacks to thread. Creates ``Queue``
implicitly if not provided.
burst_limit (:obj:`int`, optional): Number of maximum callbacks to process per time-window
defined by :attr:`time_limit_ms`. Defaults to 30.
time_limit_ms (:obj:`int`, optional): Defines width of time-window used when each
processing limit is calculated. Defaults to 1000.
exc_route (:obj:`callable`, optional): A callable, accepting 1 positional argument; used to
route exceptions from processor thread to main thread; is called on `Exception`
subclass exceptions. If not provided, exceptions are routed through dummy handler,
which re-raises them.
autostart (:obj:`bool`, optional): If :obj:`True`, processor is started immediately after
object's creation; if :obj:`False`, should be started manually by `start` method.
Defaults to :obj:`True`.
name (:obj:`str`, optional): Thread's name. Defaults to ``'DelayQueue-N'``, where N is
sequential number of object created.
Attributes:
burst_limit (:obj:`int`): Number of maximum callbacks to process per time-window.
time_limit (:obj:`int`): Defines width of time-window used when each processing limit is
calculated.
exc_route (:obj:`callable`): A callable, accepting 1 positional argument; used to route
exceptions from processor thread to main thread;
name (:obj:`str`): Thread's name.
"""
_instcnt = 0 # instance counter
def __init__(
self,
queue: q.Queue = None,
burst_limit: int = 30,
time_limit_ms: int = 1000,
exc_route: Callable[[Exception], None] = None,
autostart: bool = True,
name: str = None,
):
warnings.warn(
'DelayQueue in its current form is deprecated and will be reinvented in a future '
'release. See https://git.io/JtDbF for a list of known bugs.',
category=TelegramDeprecationWarning,
)
self._queue = queue if queue is not None else q.Queue()
self.burst_limit = burst_limit
self.time_limit = time_limit_ms / 1000
self.exc_route = exc_route if exc_route is not None else self._default_exception_handler
self.__exit_req = False # flag to gently exit thread
self.__class__._instcnt += 1
if name is None:
name = f'{self.__class__.__name__}-{self.__class__._instcnt}'
super().__init__(name=name)
self.daemon = False
if autostart: # immediately start processing
super().start()
def run(self) -> None:
"""
Do not use the method except for unthreaded testing purposes, the method normally is
automatically called by autostart argument.
"""
times: List[float] = [] # used to store each callable processing time
while True:
item = self._queue.get()
if self.__exit_req:
return # shutdown thread
# delay routine
now = time.perf_counter()
t_delta = now - self.time_limit # calculate early to improve perf.
if times and t_delta > times[-1]:
# if last call was before the limit time-window
# used to impr. perf. in long-interval calls case
times = [now]
else:
# collect last in current limit time-window
times = [t for t in times if t >= t_delta]
times.append(now)
if len(times) >= self.burst_limit: # if throughput limit was hit
time.sleep(times[1] - t_delta)
# finally process one
try:
func, args, kwargs = item
func(*args, **kwargs)
except Exception as exc: # re-route any exceptions
self.exc_route(exc) # to prevent thread exit
def stop(self, timeout: float = None) -> None:
"""Used to gently stop processor and shutdown its thread.
Args:
timeout (:obj:`float`): Indicates maximum time to wait for processor to stop and its
thread to exit. If timeout exceeds and processor has not stopped, method silently
returns. :attr:`is_alive` could be used afterwards to check the actual status.
``timeout`` set to :obj:`None`, blocks until processor is shut down.
Defaults to :obj:`None`.
"""
self.__exit_req = True # gently request
self._queue.put(None) # put something to unfreeze if frozen
super().join(timeout=timeout)
@staticmethod
def _default_exception_handler(exc: Exception) -> NoReturn:
"""
Dummy exception handler which re-raises exception in thread. Could be possibly overwritten
by subclasses.
"""
raise exc
def __call__(self, func: Callable, *args: object, **kwargs: object) -> None:
"""Used to process callbacks in throughput-limiting thread through queue.
Args:
func (:obj:`callable`): The actual function (or any callable) that is processed through
queue.
*args (:obj:`list`): Variable-length `func` arguments.
**kwargs (:obj:`dict`): Arbitrary keyword-arguments to `func`.
"""
if not self.is_alive() or self.__exit_req:
raise DelayQueueError('Could not process callback in stopped thread')
self._queue.put((func, args, kwargs))
# The most straightforward way to implement this is to use 2 sequential delay
# queues, like on classic delay chain schematics in electronics.
# So, message path is:
# msg --> group delay if group msg, else no delay --> normal msg delay --> out
# This way OS threading scheduler cares of timings accuracy.
# (see time.time, time.clock, time.perf_counter, time.sleep @ docs.python.org)
class MessageQueue:
"""
Implements callback processing with proper delays to avoid hitting Telegram's message limits.
Contains two ``DelayQueue``, for group and for all messages, interconnected in delay chain.
Callables are processed through *group* ``DelayQueue``, then through *all* ``DelayQueue`` for
group-type messages. For non-group messages, only the *all* ``DelayQueue`` is used.
.. deprecated:: 13.3
:class:`telegram.ext.MessageQueue` in its current form is deprecated and will be reinvented
in a future release. See `this thread <https://git.io/JtDbF>`_ for a list of known bugs.
Args:
all_burst_limit (:obj:`int`, optional): Number of maximum *all-type* callbacks to process
per time-window defined by :attr:`all_time_limit_ms`. Defaults to 30.
all_time_limit_ms (:obj:`int`, optional): Defines width of *all-type* time-window used when
each processing limit is calculated. Defaults to 1000 ms.
group_burst_limit (:obj:`int`, optional): Number of maximum *group-type* callbacks to
process per time-window defined by :attr:`group_time_limit_ms`. Defaults to 20.
group_time_limit_ms (:obj:`int`, optional): Defines width of *group-type* time-window used
when each processing limit is calculated. Defaults to 60000 ms.
exc_route (:obj:`callable`, optional): A callable, accepting one positional argument; used
to route exceptions from processor threads to main thread; is called on ``Exception``
subclass exceptions. If not provided, exceptions are routed through dummy handler,
which re-raises them.
autostart (:obj:`bool`, optional): If :obj:`True`, processors are started immediately after
object's creation; if :obj:`False`, should be started manually by :attr:`start` method.
Defaults to :obj:`True`.
"""
def __init__(
self,
all_burst_limit: int = 30,
all_time_limit_ms: int = 1000,
group_burst_limit: int = 20,
group_time_limit_ms: int = 60000,
exc_route: Callable[[Exception], None] = None,
autostart: bool = True,
):
warnings.warn(
'MessageQueue in its current form is deprecated and will be reinvented in a future '
'release. See https://git.io/JtDbF for a list of known bugs.',
category=TelegramDeprecationWarning,
)
# create according delay queues, use composition
self._all_delayq = DelayQueue(
burst_limit=all_burst_limit,
time_limit_ms=all_time_limit_ms,
exc_route=exc_route,
autostart=autostart,
)
self._group_delayq = DelayQueue(
burst_limit=group_burst_limit,
time_limit_ms=group_time_limit_ms,
exc_route=exc_route,
autostart=autostart,
)
def start(self) -> None:
"""Method is used to manually start the ``MessageQueue`` processing."""
self._all_delayq.start()
self._group_delayq.start()
def stop(self, timeout: float = None) -> None:
"""Stops the ``MessageQueue``."""
self._group_delayq.stop(timeout=timeout)
self._all_delayq.stop(timeout=timeout)
stop.__doc__ = DelayQueue.stop.__doc__ or '' # reuse docstring if any
def __call__(self, promise: Callable, is_group_msg: bool = False) -> Callable:
"""
Processes callables in throughput-limiting queues to avoid hitting limits (specified with
:attr:`burst_limit` and :attr:`time_limit`.
Args:
promise (:obj:`callable`): Mainly the ``telegram.utils.promise.Promise`` (see Notes for
other callables), that is processed in delay queues.
is_group_msg (:obj:`bool`, optional): Defines whether ``promise`` would be processed in
group*+*all* ``DelayQueue``s (if set to :obj:`True`), or only through *all*
``DelayQueue`` (if set to :obj:`False`), resulting in needed delays to avoid
hitting specified limits. Defaults to :obj:`False`.
Note:
Method is designed to accept ``telegram.utils.promise.Promise`` as ``promise``
argument, but other callables could be used too. For example, lambdas or simple
functions could be used to wrap original func to be called with needed args. In that
case, be sure that either wrapper func does not raise outside exceptions or the proper
:attr:`exc_route` handler is provided.
Returns:
:obj:`callable`: Used as ``promise`` argument.
"""
if not is_group_msg: # ignore middle group delay
self._all_delayq(promise)
else: # use middle group delay
self._group_delayq(self._all_delayq, promise)
return promise
def queuedmessage(method: Callable) -> Callable:
"""A decorator to be used with :attr:`telegram.Bot` send* methods.
Note:
As it probably wouldn't be a good idea to make this decorator a property, it has been coded
as decorator function, so it implies that first positional argument to wrapped MUST be
self.
The next object attributes are used by decorator:
Attributes:
self._is_messages_queued_default (:obj:`bool`): Value to provide class-defaults to
``queued`` kwarg if not provided during wrapped method call.
self._msg_queue (:class:`telegram.ext.messagequeue.MessageQueue`): The actual
``MessageQueue`` used to delay outbound messages according to specified time-limits.
Wrapped method starts accepting the next kwargs:
Args:
queued (:obj:`bool`, optional): If set to :obj:`True`, the ``MessageQueue`` is used to
process output messages. Defaults to `self._is_queued_out`.
isgroup (:obj:`bool`, optional): If set to :obj:`True`, the message is meant to be
group-type(as there's no obvious way to determine its type in other way at the moment).
Group-type messages could have additional processing delay according to limits set
in `self._out_queue`. Defaults to :obj:`False`.
Returns:
``telegram.utils.promise.Promise``: In case call is queued or original method's return
value if it's not.
"""
@functools.wraps(method)
def wrapped(self: 'Bot', *args: object, **kwargs: object) -> object:
# pylint: disable=W0212
queued = kwargs.pop(
'queued', self._is_messages_queued_default # type: ignore[attr-defined]
)
isgroup = kwargs.pop('isgroup', False)
if queued:
prom = Promise(method, (self,) + args, kwargs)
return self._msg_queue(prom, isgroup) # type: ignore[attr-defined]
return method(self, *args, **kwargs)
return wrapped