utils.py 文件源码

def get_remote_messages(config, queue, fill=True, block=False):
    """
    Get all messages from queue without removing from it

    :return: yield raw deserialized messages
    :rtype: json
    """

    to_inject = []

    try:
        while 1:
            message = queue.get(block=False, timeout=1)

            # --------------------------------------------------------------------------
            # Try to deserialize
            # --------------------------------------------------------------------------
            # Is Pickle info?
            try:
                deserialized = loads(message.body)
            except SerializationError:
                pass

            yield deserialized

            to_inject.append(deserialized)

    except Empty:
        # When Queue is Empty -> reinject all removed messages
        if fill is True:
            for x in to_inject:
                queue.put(x, serializer="pickle")


# ----------------------------------------------------------------------