def get_replay(name, query, config, context=None):
    """
    Query the replay endpoint for missed messages.

    Args:
        name (str): The replay endpoint name.
        query (dict): A dictionary used to query the replay endpoint for messages.
            Queries are dictionaries with the following any of the following keys:

            * 'seq_ids': A ``list`` of ``int``, matching the seq_id attributes
              of the messages. It should return at most as many messages as the
              length of the list, assuming no duplicate.

            * 'seq_id': A single ``int`` matching the seq_id attribute of the message.
              Should return a single message. It is intended as a shorthand for
              singleton ``seq_ids`` queries.

            * 'seq_id_range': A two-tuple of ``int`` defining a range of seq_id to check.

            * 'msg_ids': A ``list`` of UUIDs matching the msg_id attribute of the messages.

            * 'msg_id': A single UUID for the msg_id attribute.

            * 'time': A tuple of two timestamps. It will return all messages emitted in between.
        config (dict): A configuration dictionary. This dictionary should contain, at a
            minimum, two keys. The first key, 'replay_endpoints', should be a dictionary
            that maps ``name`` to a ZeroMQ socket. The second key, 'io_threads', is an
            integer used to initialize the ZeroMQ context.
        context (zmq.Context): The ZeroMQ context to use. If a context is not provided,
            one will be created.

    Returns:
        generator: A generator that yields message dictionaries.
    """
    endpoint = config.get('replay_endpoints', {}).get(name, None)
    if not endpoint:
        raise IOError("No appropriate replay endpoint "
                      "found for {0}".format(name))

    if not context:
        context = zmq.Context(config['io_threads'])

    # A replay endpoint isn't PUB/SUB but REQ/REP, as it allows
    # for bidirectional communication
    socket = context.socket(zmq.REQ)
    try:
        socket.connect(endpoint)
    except zmq.ZMQError as e:
        raise IOError("Error when connecting to the "
                      "replay endpoint: '{0}'".format(str(e)))

    # REQ/REP dance
    socket.send(fedmsg.encoding.dumps(query).encode('utf-8'))
    msgs = socket.recv_multipart()
    socket.close()

    for m in msgs:
        try:
            yield fedmsg.encoding.loads(m.decode('utf-8'))
        except ValueError:
            # We assume that if it isn't JSON then it's an error message
            raise ValueError(m)
def get_replay ( name , query , config , context = None ) : endpoint = config . get ( 'replay_endpoints' , { } ) . get ( name , None ) if not endpoint : raise IOError ( "No appropriate replay endpoint " "found for {0}" . format ( name ) ) if not context : context = zmq . Context ( config [ 'io_threads' ] ) socket = context . socket ( zmq . REQ ) try : socket . connect ( endpoint ) except zmq . ZMQError as e : raise IOError ( "Error when connecting to the " "replay endpoint: '{0}'" . format ( str ( e ) ) ) socket . send ( fedmsg . encoding . dumps ( query ) . encode ( 'utf-8' ) ) msgs = socket . recv_multipart ( ) socket . close ( ) for m in msgs : try : yield fedmsg . encoding . loads ( m . decode ( 'utf-8' ) ) except ValueError : raise ValueError ( m )
Query the replay endpoint for missed messages.

    Args:
        name (str): The replay endpoint name.
        query (dict): A dictionary used to query the replay endpoint for messages.
            Queries are dictionaries with the following any of the following keys:

            * 'seq_ids': A ``list`` of ``int``, matching the seq_id attributes
              of the messages. It should return at most as many messages as the
              length of the list, assuming no duplicate.

            * 'seq_id': A single ``int`` matching the seq_id attribute of the message.
              Should return a single message. It is intended as a shorthand for
              singleton ``seq_ids`` queries.

            * 'seq_id_range': A two-tuple of ``int`` defining a range of seq_id to check.

            * 'msg_ids': A ``list`` of UUIDs matching the msg_id attribute of the messages.

            * 'msg_id': A single UUID for the msg_id attribute.

            * 'time': A tuple of two timestamps. It will return all messages emitted in between.
        config (dict): A configuration dictionary. This dictionary should contain, at a
            minimum, two keys. The first key, 'replay_endpoints', should be a dictionary
            that maps ``name`` to a ZeroMQ socket. The second key, 'io_threads', is an
            integer used to initialize the ZeroMQ context.
        context (zmq.Context): The ZeroMQ context to use. If a context is not provided,
            one will be created.

    Returns:
        generator: A generator that yields message dictionaries.