'Get rows to be sent over federation between the two tokens
Args:
from_token (int)
to_token(int)
limit (int)
federation_ack (int): Optional. The position where the worker is
explicitly acknowledged it has handled. Allows us to drop
data from before that point'

def get_replication_rows(self, from_token, to_token, limit, federation_ack=None):

    if (from_token > self.pos):
        from_token = (-1)
    rows = []
    if federation_ack:
        self._clear_queue_before_pos(federation_ack)
    keys = self.presence_changed.keys()
    i = keys.bisect_right(from_token)
    j = (keys.bisect_right(to_token) + 1)
    dest_user_ids = [(pos, user_id) for pos in keys[i:j] for user_id in self.presence_changed[pos]]
    for (key, user_id) in dest_user_ids:
        rows.append((key, PresenceRow(state=self.presence_map[user_id])))
    keys = self.keyed_edu_changed.keys()
    i = keys.bisect_right(from_token)
    j = (keys.bisect_right(to_token) + 1)
    keyed_edus = {self.keyed_edu_changed[k]: k for k in keys[i:j]}
    for ((destination, edu_key), pos) in keyed_edus.iteritems():
        rows.append((pos, KeyedEduRow(key=edu_key, edu=self.keyed_edu[(destination, edu_key)])))
    keys = self.edus.keys()
    i = keys.bisect_right(from_token)
    j = (keys.bisect_right(to_token) + 1)
    edus = ((k, self.edus[k]) for k in keys[i:j])
    for (pos, edu) in edus:
        rows.append((pos, EduRow(edu)))
    keys = self.failures.keys()
    i = keys.bisect_right(from_token)
    j = (keys.bisect_right(to_token) + 1)
    failures = ((k, self.failures[k]) for k in keys[i:j])
    for (pos, (destination, failure)) in failures:
        rows.append((pos, FailureRow(destination=destination, failure=failure)))
    keys = self.device_messages.keys()
    i = keys.bisect_right(from_token)
    j = (keys.bisect_right(to_token) + 1)
    device_messages = {self.device_messages[k]: k for k in keys[i:j]}
    for (destination, pos) in device_messages.iteritems():
        rows.append((pos, DeviceRow(destination=destination)))
    rows.sort()
    return [(pos, row.TypeId, row.to_data()) for (pos, row) in rows]
