synapse/federation/federation_server.py
def _handle_incoming_transaction(self, transaction, request_time):
        """ Process an incoming transaction and return the HTTP response

        Args:
            transaction (Transaction): incoming transaction
            request_time (int): timestamp that the HTTP request arrived at

        Returns:
            Deferred[(int, object)]: http response code and body
        """
        response = yield self.transaction_actions.have_responded(transaction)

        if response:
            logger.debug(
                "[%s] We've already responded to this request",
                transaction.transaction_id
            )
            defer.returnValue(response)
            return

        logger.debug("[%s] Transaction is new", transaction.transaction_id)

        received_pdus_counter.inc_by(len(transaction.pdus))

        pdus_by_room = {}

        for p in transaction.pdus:
            if "unsigned" in p:
                unsigned = p["unsigned"]
                if "age" in unsigned:
                    p["age"] = unsigned["age"]
            if "age" in p:
                p["age_ts"] = request_time - int(p["age"])
                del p["age"]

            event = event_from_pdu_json(p)
            room_id = event.room_id
            pdus_by_room.setdefault(room_id, []).append(event)

        pdu_results = {}

        # we can process different rooms in parallel (which is useful if they
        # require callouts to other servers to fetch missing events), but
        # impose a limit to avoid going too crazy with ram/cpu.
        @defer.inlineCallbacks
        def process_pdus_for_room(room_id):
            logger.debug("Processing PDUs for %s", room_id)
            for pdu in pdus_by_room[room_id]:
                event_id = pdu.event_id
                try:
                    yield self._handle_received_pdu(
                        transaction.origin, pdu
                    )
                    pdu_results[event_id] = {}
                except FederationError as e:
                    logger.warn("Error handling PDU %s: %s", event_id, e)
                    pdu_results[event_id] = {"error": str(e)}
                except Exception as e:
                    pdu_results[event_id] = {"error": str(e)}
                    logger.exception("Failed to handle PDU %s", event_id)

        yield async.concurrently_execute(
            process_pdus_for_room, pdus_by_room.keys(),
            TRANSACTION_CONCURRENCY_LIMIT,
        )

        if hasattr(transaction, "edus"):
            for edu in (Edu(**x) for x in transaction.edus):
                yield self.received_edu(
                    transaction.origin,
                    edu.edu_type,
                    edu.content
                )

        pdu_failures = getattr(transaction, "pdu_failures", [])
        for failure in pdu_failures:
            logger.info("Got failure %r", failure)

        response = {
            "pdus": pdu_results,
        }

        logger.debug("Returning: %s", str(response))

        yield self.transaction_actions.set_response(
            transaction,
            200, response
        )
        defer.returnValue((200, response))
 Process an incoming transaction and return the HTTP response

        Args:
            transaction (Transaction): incoming transaction
            request_time (int): timestamp that the HTTP request arrived at

        Returns:
            Deferred[(int, object)]: http response code and body
        
Process an incoming transaction and return the HTTP response
Process an incoming transaction and return the HTTP response
def _handle_incoming_transaction(self, transaction, request_time):
        response = yield self.transaction_actions.have_responded(transaction)
        if response:
            logger.debug(
                "[%s] We've already responded to this request",
                transaction.transaction_id
            )
            defer.returnValue(response)
            return
        logger.debug("[%s] Transaction is new", transaction.transaction_id)
        received_pdus_counter.inc_by(len(transaction.pdus))
        pdus_by_room = {}
        for p in transaction.pdus:
            if "unsigned" in p:
                unsigned = p["unsigned"]
                if "age" in unsigned:
                    p["age"] = unsigned["age"]
            if "age" in p:
                p["age_ts"] = request_time - int(p["age"])
                del p["age"]
            event = event_from_pdu_json(p)
            room_id = event.room_id
            pdus_by_room.setdefault(room_id, []).append(event)
        pdu_results = {}
        @defer.inlineCallbacks
        def process_pdus_for_room(room_id):
            logger.debug("Processing PDUs for %s", room_id)
            for pdu in pdus_by_room[room_id]:
                event_id = pdu.event_id
                try:
                    yield self._handle_received_pdu(
                        transaction.origin, pdu
                    )
                    pdu_results[event_id] = {}
                except FederationError as e:
                    logger.warn("Error handling PDU %s: %s", event_id, e)
                    pdu_results[event_id] = {"error": str(e)}
                except Exception as e:
                    pdu_results[event_id] = {"error": str(e)}
                    logger.exception("Failed to handle PDU %s", event_id)
        yield async.concurrently_execute(
            process_pdus_for_room, pdus_by_room.keys(),
            TRANSACTION_CONCURRENCY_LIMIT,
        )
        if hasattr(transaction, "edus"):
            for edu in (Edu(**x) for x in transaction.edus):
                yield self.received_edu(
                    transaction.origin,
                    edu.edu_type,
                    edu.content
                )
        pdu_failures = getattr(transaction, "pdu_failures", [])
        for failure in pdu_failures:
            logger.info("Got failure %r", failure)
        response = {
            "pdus": pdu_results,
        }
        logger.debug("Returning: %s", str(response))
        yield self.transaction_actions.set_response(
            transaction,
            200, response
        )
        defer.returnValue((200, response))
def _handle_incoming_transaction ( self , transaction , request_time ) : response = yield self . transaction_actions . have_responded ( transaction ) if response : logger . debug ( "[%s] We've already responded to this request" , transaction . transaction_id ) defer . returnValue ( response ) return logger . debug ( "[%s] Transaction is new" , transaction . transaction_id ) received_pdus_counter . inc_by ( len ( transaction . pdus ) ) pdus_by_room = { } for p in transaction . pdus : if "unsigned" in p : unsigned = p [ "unsigned" ] if "age" in unsigned : p [ "age" ] = unsigned [ "age" ] if "age" in p : p [ "age_ts" ] = request_time - int ( p [ "age" ] ) del p [ "age" ] event = event_from_pdu_json ( p ) room_id = event . room_id pdus_by_room . setdefault ( room_id , [ ] ) . append ( event ) pdu_results = { } @ defer . inlineCallbacks def process_pdus_for_room ( room_id ) : logger . debug ( "Processing PDUs for %s" , room_id ) for pdu in pdus_by_room [ room_id ] : event_id = pdu . event_id try : yield self . _handle_received_pdu ( transaction . origin , pdu ) pdu_results [ event_id ] = { } except FederationError as e : logger . warn ( "Error handling PDU %s: %s" , event_id , e ) pdu_results [ event_id ] = { "error" : str ( e ) } except Exception as e : pdu_results [ event_id ] = { "error" : str ( e ) } logger . exception ( "Failed to handle PDU %s" , event_id ) yield async . concurrently_execute ( process_pdus_for_room , pdus_by_room . keys ( ) , TRANSACTION_CONCURRENCY_LIMIT , ) if hasattr ( transaction , "edus" ) : for edu in ( Edu ( ** x ) for x in transaction . edus ) : yield self . received_edu ( transaction . origin , edu . edu_type , edu . content ) pdu_failures = getattr ( transaction , "pdu_failures" , [ ] ) for failure in pdu_failures : logger . info ( "Got failure %r" , failure ) response = { "pdus" : pdu_results , } logger . debug ( "Returning: %s" , str ( response ) ) yield self . transaction_actions . set_response ( transaction , 200 , response ) defer . returnValue ( ( 200 , response ) )
Process an incoming transaction and return the HTTP response
Process an incoming transaction and return the HTTP response
""" Process an incoming transaction and return the HTTP response

        Args:
            transaction (Transaction): incoming transaction
            request_time (int): timestamp that the HTTP request arrived at

        Returns:
            Deferred[(int, object)]: http response code and body
        """ # we can process different rooms in parallel (which is useful if they # require callouts to other servers to fetch missing events), but # impose a limit to avoid going too crazy with ram/cpu.
http response code and body
http response code and body
Deferred[(int, object)]
self
transaction
None
request_time
timestamp that the HTTP request arrived at
timestamp that the HTTP request arrived at