synapse/federation/send_queue.py
def process_rows_for_federation(transaction_queue, rows):
    """Parse a list of rows from the federation stream and put them in the
    transaction queue ready for sending to the relevant homeservers.

    Args:
        transaction_queue (TransactionQueue)
        rows (list(synapse.replication.tcp.streams.FederationStreamRow))
    """

    # The federation stream contains a bunch of different types of
    # rows that need to be handled differently. We parse the rows, put
    # them into the appropriate collection and then send them off.

    buff = ParsedFederationStreamData(
        presence=[],
        keyed_edus={},
        edus={},
        failures={},
        device_destinations=set(),
    )

    # Parse the rows in the stream and add to the buffer
    for row in rows:
        if row.type not in TypeToRow:
            logger.error("Unrecognized federation row type %r", row.type)
            continue

        RowType = TypeToRow[row.type]
        parsed_row = RowType.from_data(row.data)
        parsed_row.add_to_buffer(buff)

    if buff.presence:
        transaction_queue.send_presence(buff.presence)

    for destination, edu_map in iteritems(buff.keyed_edus):
        for key, edu in edu_map.items():
            transaction_queue.send_edu(
                edu.destination, edu.edu_type, edu.content, key=key,
            )

    for destination, edu_list in iteritems(buff.edus):
        for edu in edu_list:
            transaction_queue.send_edu(
                edu.destination, edu.edu_type, edu.content, key=None,
            )

    for destination, failure_list in iteritems(buff.failures):
        for failure in failure_list:
            transaction_queue.send_failure(destination, failure)

    for destination in buff.device_destinations:
        transaction_queue.send_device_messages(destination)
Parse a list of rows from the federation stream and put them in the
    transaction queue ready for sending to the relevant homeservers.

    Args:
        transaction_queue (TransactionQueue)
        rows (list(synapse.replication.tcp.streams.FederationStreamRow))
    
Parse a list of rows from the federation stream and put them in the
transaction queue ready for sending to the relevant homeservers.


Parse a list of rows from the federation stream and put them in the transaction queue ready for sending to the relevant homeservers .
def process_rows_for_federation(transaction_queue, rows):
    buff = ParsedFederationStreamData(
        presence=[],
        keyed_edus={},
        edus={},
        failures={},
        device_destinations=set(),
    )
    for row in rows:
        if row.type not in TypeToRow:
            logger.error("Unrecognized federation row type %r", row.type)
            continue
        RowType = TypeToRow[row.type]
        parsed_row = RowType.from_data(row.data)
        parsed_row.add_to_buffer(buff)
    if buff.presence:
        transaction_queue.send_presence(buff.presence)
    for destination, edu_map in iteritems(buff.keyed_edus):
        for key, edu in edu_map.items():
            transaction_queue.send_edu(
                edu.destination, edu.edu_type, edu.content, key=key,
            )
    for destination, edu_list in iteritems(buff.edus):
        for edu in edu_list:
            transaction_queue.send_edu(
                edu.destination, edu.edu_type, edu.content, key=None,
            )
    for destination, failure_list in iteritems(buff.failures):
        for failure in failure_list:
            transaction_queue.send_failure(destination, failure)
    for destination in buff.device_destinations:
        transaction_queue.send_device_messages(destination)
def process_rows_for_federation ( transaction_queue , rows ) : buff = ParsedFederationStreamData ( presence = [ ] , keyed_edus = { } , edus = { } , failures = { } , device_destinations = set ( ) , ) for row in rows : if row . type not in TypeToRow : logger . error ( "Unrecognized federation row type %r" , row . type ) continue RowType = TypeToRow [ row . type ] parsed_row = RowType . from_data ( row . data ) parsed_row . add_to_buffer ( buff ) if buff . presence : transaction_queue . send_presence ( buff . presence ) for destination , edu_map in iteritems ( buff . keyed_edus ) : for key , edu in edu_map . items ( ) : transaction_queue . send_edu ( edu . destination , edu . edu_type , edu . content , key = key , ) for destination , edu_list in iteritems ( buff . edus ) : for edu in edu_list : transaction_queue . send_edu ( edu . destination , edu . edu_type , edu . content , key = None , ) for destination , failure_list in iteritems ( buff . failures ) : for failure in failure_list : transaction_queue . send_failure ( destination , failure ) for destination in buff . device_destinations : transaction_queue . send_device_messages ( destination )
Parse a list of rows from the federation stream and put them in the
transaction queue ready for sending to the relevant homeservers.
Parse a list of rows from the federation stream and put them in the transaction queue ready for sending to the relevant homeservers .
"""Parse a list of rows from the federation stream and put them in the
    transaction queue ready for sending to the relevant homeservers.

    Args:
        transaction_queue (TransactionQueue)
        rows (list(synapse.replication.tcp.streams.FederationStreamRow))
    """ # The federation stream contains a bunch of different types of # rows that need to be handled differently. We parse the rows, put # them into the appropriate collection and then send them off. # Parse the rows in the stream and add to the buffer
transaction_queue
rows