def message_loop(self, callback=None, relax=0.1,
                     timeout=20, allowed_updates=None,
                     source=None, ordered=True, maxhold=3,
                     run_forever=False):
        """
        :deprecated: will be removed in future. Use :class:`.MessageLoop` instead.

        Spawn a thread to constantly ``getUpdates`` or pull updates from a queue.
        Apply ``callback`` to every message received. Also starts the scheduler thread
        for internal events.

        :param callback:
            a function that takes one argument (the message), or a routing table.
            If ``None``, the bot's ``handle`` method is used.

        A *routing table* is a dictionary of ``{flavor: function}``, mapping messages to appropriate
        handler functions according to their flavors. It allows you to define functions specifically
        to handle one flavor of messages. It usually looks like this: ``{'chat': fn1,
        'callback_query': fn2, 'inline_query': fn3, ...}``. Each handler function should take
        one argument (the message).



        :param source:
            Source of updates.
            If ``None``, ``getUpdates`` is used to obtain new messages from Telegram servers.
            If it is a synchronized queue (``Queue.Queue`` in Python 2.7 or
            ``queue.Queue`` in Python 3), new messages are pulled from the queue.
            A web application implementing a webhook can dump updates into the queue,
            while the bot pulls from it. This is how amanobot can be integrated with webhooks.

        Acceptable contents in queue:

        - ``str``, ``unicode`` (Python 2.7), or ``bytes`` (Python 3, decoded using UTF-8)
          representing a JSON-serialized `Update <https://core.telegram.org/bots/api#update>`_ object.
        - a ``dict`` representing an Update object.

        When ``source`` is ``None``, these parameters are meaningful:

        :type relax: float
        :param relax: seconds between each ``getUpdates``

        :type timeout: int
        :param timeout:
            ``timeout`` parameter supplied to :meth:`amanobot.Bot.getUpdates`,
            controlling how long to poll.

        :type allowed_updates: array of string
        :param allowed_updates:
            ``allowed_updates`` parameter supplied to :meth:`amanobot.Bot.getUpdates`,
            controlling which types of updates to receive.

        When ``source`` is a queue, these parameters are meaningful:

        :type ordered: bool
        :param ordered:
            If ``True``, ensure in-order delivery of messages to ``callback``
            (i.e. updates with a smaller ``update_id`` always come before those with
            a larger ``update_id``).
            If ``False``, no re-ordering is done. ``callback`` is applied to messages
            as soon as they are pulled from queue.

        :type maxhold: float
        :param maxhold:
            Applied only when ``ordered`` is ``True``. The maximum number of seconds
            an update is held waiting for a not-yet-arrived smaller ``update_id``.
            When this number of seconds is up, the update is delivered to ``callback``
            even if some smaller ``update_id``\s have not yet arrived. If those smaller
            ``update_id``\s arrive at some later time, they are discarded.

        Finally, there is this parameter, meaningful always:

        :type run_forever: bool or str
        :param run_forever:
            If ``True`` or any non-empty string, append an infinite loop at the end of
            this method, so it never returns. Useful as the very last line in a program.
            A non-empty string will also be printed, useful as an indication that the
            program is listening.
        """
        if callback is None:
            callback = self.handle
        elif isinstance(callback, dict):
            callback = flavor_router(callback)

        collect_queue = queue.Queue()

        def collector():
            while 1:
                try:
                    item = collect_queue.get(block=True)
                    callback(item)
                except:
                    # Localize error so thread can keep going.
                    traceback.print_exc()

        def relay_to_collector(update):
            key = _find_first_key(update, ['message',
                                           'edited_message',
                                           'poll',
                                           'channel_post',
                                           'edited_channel_post',
                                           'callback_query',
                                           'inline_query',
                                           'chosen_inline_result',
                                           'shipping_query',
                                           'pre_checkout_query'])
            collect_queue.put(update[key])
            return update['update_id']

        def get_from_telegram_server():
            offset = None  # running offset
            allowed_upd = allowed_updates
            while 1:
                try:
                    result = self.getUpdates(offset=offset,
                                             timeout=timeout,
                                             allowed_updates=allowed_upd)

                    # Once passed, this parameter is no longer needed.
                    allowed_upd = None

                    if len(result) > 0:
                        # No sort. Trust server to give messages in correct order.
                        # Update offset to max(update_id) + 1
                        offset = max([relay_to_collector(update) for update in result]) + 1

                except exception.BadHTTPResponse as e:
                    traceback.print_exc()

                    # Servers probably down. Wait longer.
                    if e.status == 502:
                        time.sleep(30)
                except:
                    traceback.print_exc()
                finally:
                    time.sleep(relax)

        def dictify3(data):
            if type(data) is bytes:
                return json.loads(data.decode('utf-8'))
            elif type(data) is str:
                return json.loads(data)
            elif type(data) is dict:
                return data
            else:
                raise ValueError()

        def dictify27(data):
            if type(data) in [str, unicode]:
                return json.loads(data)
            elif type(data) is dict:
                return data
            else:
                raise ValueError()

        def get_from_queue_unordered(qu):
            dictify = dictify3 if sys.version_info >= (3,) else dictify27
            while 1:
                try:
                    data = qu.get(block=True)
                    update = dictify(data)
                    relay_to_collector(update)
                except:
                    traceback.print_exc()

        def get_from_queue(qu):
            dictify = dictify3 if sys.version_info >= (3,) else dictify27

            # Here is the re-ordering mechanism, ensuring in-order delivery of updates.
            max_id = None                 # max update_id passed to callback
            buffer = collections.deque()  # keep those updates which skip some update_id
            qwait = None                  # how long to wait for updates,
                                          # because buffer's content has to be returned in time.

            while 1:
                try:
                    data = qu.get(block=True, timeout=qwait)
                    update = dictify(data)

                    if max_id is None:
                        # First message received, handle regardless.
                        max_id = relay_to_collector(update)

                    elif update['update_id'] == max_id + 1:
                        # No update_id skipped, handle naturally.
                        max_id = relay_to_collector(update)

                        # clear contagious updates in buffer
                        if len(buffer) > 0:
                            buffer.popleft()  # first element belongs to update just received, useless now.
                            while 1:
                                try:
                                    if type(buffer[0]) is dict:
                                        max_id = relay_to_collector(buffer.popleft())  # updates that arrived earlier, handle them.
                                    else:
                                        break  # gap, no more contagious updates
                                except IndexError:
                                    break  # buffer empty

                    elif update['update_id'] > max_id + 1:
                        # Update arrives pre-maturely, insert to buffer.
                        nbuf = len(buffer)
                        if update['update_id'] <= max_id + nbuf:
                            # buffer long enough, put update at position
                            buffer[update['update_id'] - max_id - 1] = update
                        else:
                            # buffer too short, lengthen it
                            expire = time.time() + maxhold
                            for a in range(nbuf, update['update_id']-max_id-1):
                                buffer.append(expire)  # put expiry time in gaps
                            buffer.append(update)

                    else:
                        pass  # discard

                except queue.Empty:
                    # debug message
                    # print('Timeout')

                    # some buffer contents have to be handled
                    # flush buffer until a non-expired time is encountered
                    while 1:
                        try:
                            if type(buffer[0]) is dict:
                                max_id = relay_to_collector(buffer.popleft())
                            else:
                                expire = buffer[0]
                                if expire <= time.time():
                                    max_id += 1
                                    buffer.popleft()
                                else:
                                    break  # non-expired
                        except IndexError:
                            break  # buffer empty
                except:
                    traceback.print_exc()
                finally:
                    try:
                        # don't wait longer than next expiry time
                        qwait = buffer[0] - time.time()
                        if qwait < 0:
                            qwait = 0
                    except IndexError:
                        # buffer empty, can wait forever
                        qwait = None

                    # debug message
                    # print ('Buffer:', str(buffer), ', To Wait:', qwait, ', Max ID:', max_id)

        collector_thread = threading.Thread(target=collector)
        collector_thread.daemon = True
        collector_thread.start()

        if source is None:
            message_thread = threading.Thread(target=get_from_telegram_server)
        elif isinstance(source, queue.Queue):
            if ordered:
                message_thread = threading.Thread(target=get_from_queue, args=(source,))
            else:
                message_thread = threading.Thread(target=get_from_queue_unordered, args=(source,))
        else:
            raise ValueError('Invalid source')

        message_thread.daemon = True  # need this for main thread to be killable by Ctrl-C
        message_thread.start()

        self._scheduler.on_event(collect_queue.put)
        self._scheduler.run_as_thread()

        if run_forever:
            if _isstring(run_forever):
                print(run_forever)
            while 1:
                time.sleep(10)
def message_loop ( self , callback = None , relax = 0.1 , timeout = 20 , allowed_updates = None , source = None , ordered = True , maxhold = 3 , run_forever = False ) : if callback is None : callback = self . handle elif isinstance ( callback , dict ) : callback = flavor_router ( callback ) collect_queue = queue . Queue ( ) def collector ( ) : while 1 : try : item = collect_queue . get ( block = True ) callback ( item ) except : traceback . print_exc ( ) def relay_to_collector ( update ) : key = _find_first_key ( update , [ 'message' , 'edited_message' , 'poll' , 'channel_post' , 'edited_channel_post' , 'callback_query' , 'inline_query' , 'chosen_inline_result' , 'shipping_query' , 'pre_checkout_query' ] ) collect_queue . put ( update [ key ] ) return update [ 'update_id' ] def get_from_telegram_server ( ) : offset = None allowed_upd = allowed_updates while 1 : try : result = self . getUpdates ( offset = offset , timeout = timeout , allowed_updates = allowed_upd ) allowed_upd = None if len ( result ) > 0 : offset = max ( [ relay_to_collector ( update ) for update in result ] ) + 1 except exception . BadHTTPResponse as e : traceback . print_exc ( ) if e . status == 502 : time . sleep ( 30 ) except : traceback . print_exc ( ) finally : time . sleep ( relax ) def dictify3 ( data ) : if type ( data ) is bytes : return json . loads ( data . decode ( 'utf-8' ) ) elif type ( data ) is str : return json . loads ( data ) elif type ( data ) is dict : return data else : raise ValueError ( ) def dictify27 ( data ) : if type ( data ) in [ str , unicode ] : return json . loads ( data ) elif type ( data ) is dict : return data else : raise ValueError ( ) def get_from_queue_unordered ( qu ) : dictify = dictify3 if sys . version_info >= ( 3 , ) else dictify27 while 1 : try : data = qu . get ( block = True ) update = dictify ( data ) relay_to_collector ( update ) except : traceback . print_exc ( ) def get_from_queue ( qu ) : dictify = dictify3 if sys . version_info >= ( 3 , ) else dictify27 max_id = None buffer = collections . deque ( ) qwait = None while 1 : try : data = qu . get ( block = True , timeout = qwait ) update = dictify ( data ) if max_id is None : max_id = relay_to_collector ( update ) elif update [ 'update_id' ] == max_id + 1 : max_id = relay_to_collector ( update ) if len ( buffer ) > 0 : buffer . popleft ( ) while 1 : try : if type ( buffer [ 0 ] ) is dict : max_id = relay_to_collector ( buffer . popleft ( ) ) else : break except IndexError : break elif update [ 'update_id' ] > max_id + 1 : nbuf = len ( buffer ) if update [ 'update_id' ] <= max_id + nbuf : buffer [ update [ 'update_id' ] - max_id - 1 ] = update else : expire = time . time ( ) + maxhold for a in range ( nbuf , update [ 'update_id' ] - max_id - 1 ) : buffer . append ( expire ) buffer . append ( update ) else : pass except queue . Empty : while 1 : try : if type ( buffer [ 0 ] ) is dict : max_id = relay_to_collector ( buffer . popleft ( ) ) else : expire = buffer [ 0 ] if expire <= time . time ( ) : max_id += 1 buffer . popleft ( ) else : break except IndexError : break except : traceback . print_exc ( ) finally : try : qwait = buffer [ 0 ] - time . time ( ) if qwait < 0 : qwait = 0 except IndexError : qwait = None collector_thread = threading . Thread ( target = collector ) collector_thread . daemon = True collector_thread . start ( ) if source is None : message_thread = threading . Thread ( target = get_from_telegram_server ) elif isinstance ( source , queue . Queue ) : if ordered : message_thread = threading . Thread ( target = get_from_queue , args = ( source , ) ) else : message_thread = threading . Thread ( target = get_from_queue_unordered , args = ( source , ) ) else : raise ValueError ( 'Invalid source' ) message_thread . daemon = True message_thread . start ( ) self . _scheduler . on_event ( collect_queue . put ) self . _scheduler . run_as_thread ( ) if run_forever : if _isstring ( run_forever ) : print ( run_forever ) while 1 : time . sleep ( 10 )
:deprecated: will be removed in future. Use :class:`.MessageLoop` instead.

        Spawn a thread to constantly ``getUpdates`` or pull updates from a queue.
        Apply ``callback`` to every message received. Also starts the scheduler thread
        for internal events.

        :param callback:
            a function that takes one argument (the message), or a routing table.
            If ``None``, the bot's ``handle`` method is used.

        A *routing table* is a dictionary of ``{flavor: function}``, mapping messages to appropriate
        handler functions according to their flavors. It allows you to define functions specifically
        to handle one flavor of messages. It usually looks like this: ``{'chat': fn1,
        'callback_query': fn2, 'inline_query': fn3, ...}``. Each handler function should take
        one argument (the message).



        :param source:
            Source of updates.
            If ``None``, ``getUpdates`` is used to obtain new messages from Telegram servers.
            If it is a synchronized queue (``Queue.Queue`` in Python 2.7 or
            ``queue.Queue`` in Python 3), new messages are pulled from the queue.
            A web application implementing a webhook can dump updates into the queue,
            while the bot pulls from it. This is how amanobot can be integrated with webhooks.

        Acceptable contents in queue:

        - ``str``, ``unicode`` (Python 2.7), or ``bytes`` (Python 3, decoded using UTF-8)
          representing a JSON-serialized `Update <https://core.telegram.org/bots/api#update>`_ object.
        - a ``dict`` representing an Update object.

        When ``source`` is ``None``, these parameters are meaningful:

        :type relax: float
        :param relax: seconds between each ``getUpdates``

        :type timeout: int
        :param timeout:
            ``timeout`` parameter supplied to :meth:`amanobot.Bot.getUpdates`,
            controlling how long to poll.

        :type allowed_updates: array of string
        :param allowed_updates:
            ``allowed_updates`` parameter supplied to :meth:`amanobot.Bot.getUpdates`,
            controlling which types of updates to receive.

        When ``source`` is a queue, these parameters are meaningful:

        :type ordered: bool
        :param ordered:
            If ``True``, ensure in-order delivery of messages to ``callback``
            (i.e. updates with a smaller ``update_id`` always come before those with
            a larger ``update_id``).
            If ``False``, no re-ordering is done. ``callback`` is applied to messages
            as soon as they are pulled from queue.

        :type maxhold: float
        :param maxhold:
            Applied only when ``ordered`` is ``True``. The maximum number of seconds
            an update is held waiting for a not-yet-arrived smaller ``update_id``.
            When this number of seconds is up, the update is delivered to ``callback``
            even if some smaller ``update_id``\s have not yet arrived. If those smaller
            ``update_id``\s arrive at some later time, they are discarded.

        Finally, there is this parameter, meaningful always:

        :type run_forever: bool or str
        :param run_forever:
            If ``True`` or any non-empty string, append an infinite loop at the end of
            this method, so it never returns. Useful as the very last line in a program.
            A non-empty string will also be printed, useful as an indication that the
            program is listening.