synapse/storage/deviceinbox.py
def add_messages_to_device_inbox(self, local_messages_by_user_then_device,
                                     remote_messages_by_destination):
        """Used to send messages from this server.

        Args:
            sender_user_id(str): The ID of the user sending these messages.
            local_messages_by_user_and_device(dict):
                Dictionary of user_id to device_id to message.
            remote_messages_by_destination(dict):
                Dictionary of destination server_name to the EDU JSON to send.
        Returns:
            A deferred stream_id that resolves when the messages have been
            inserted.
        """

        def add_messages_txn(txn, now_ms, stream_id):
            # Add the local messages directly to the local inbox.
            self._add_messages_to_local_device_inbox_txn(
                txn, stream_id, local_messages_by_user_then_device
            )

            # Add the remote messages to the federation outbox.
            # We'll send them to a remote server when we next send a
            # federation transaction to that destination.
            sql = (
                "INSERT INTO device_federation_outbox"
                " (destination, stream_id, queued_ts, messages_json)"
                " VALUES (?,?,?,?)"
            )
            rows = []
            for destination, edu in remote_messages_by_destination.items():
                edu_json = simplejson.dumps(edu)
                rows.append((destination, stream_id, now_ms, edu_json))
            txn.executemany(sql, rows)

        with self._device_inbox_id_gen.get_next() as stream_id:
            now_ms = self.clock.time_msec()
            yield self.runInteraction(
                "add_messages_to_device_inbox",
                add_messages_txn,
                now_ms,
                stream_id,
            )
            for user_id in local_messages_by_user_then_device.keys():
                self._device_inbox_stream_cache.entity_has_changed(
                    user_id, stream_id
                )
            for destination in remote_messages_by_destination.keys():
                self._device_federation_outbox_stream_cache.entity_has_changed(
                    destination, stream_id
                )

        defer.returnValue(self._device_inbox_id_gen.get_current_token())
Used to send messages from this server.

        Args:
            sender_user_id(str): The ID of the user sending these messages.
            local_messages_by_user_and_device(dict):
                Dictionary of user_id to device_id to message.
            remote_messages_by_destination(dict):
                Dictionary of destination server_name to the EDU JSON to send.
        Returns:
            A deferred stream_id that resolves when the messages have been
            inserted.
        
Used to send messages from this server.
Used to send messages from this server .
def add_messages_to_device_inbox(self, local_messages_by_user_then_device,
                                     remote_messages_by_destination):
        def add_messages_txn(txn, now_ms, stream_id):
            self._add_messages_to_local_device_inbox_txn(
                txn, stream_id, local_messages_by_user_then_device
            )
            sql = (
                "INSERT INTO device_federation_outbox"
                " (destination, stream_id, queued_ts, messages_json)"
                " VALUES (?,?,?,?)"
            )
            rows = []
            for destination, edu in remote_messages_by_destination.items():
                edu_json = simplejson.dumps(edu)
                rows.append((destination, stream_id, now_ms, edu_json))
            txn.executemany(sql, rows)
        with self._device_inbox_id_gen.get_next() as stream_id:
            now_ms = self.clock.time_msec()
            yield self.runInteraction(
                "add_messages_to_device_inbox",
                add_messages_txn,
                now_ms,
                stream_id,
            )
            for user_id in local_messages_by_user_then_device.keys():
                self._device_inbox_stream_cache.entity_has_changed(
                    user_id, stream_id
                )
            for destination in remote_messages_by_destination.keys():
                self._device_federation_outbox_stream_cache.entity_has_changed(
                    destination, stream_id
                )
        defer.returnValue(self._device_inbox_id_gen.get_current_token())
def add_messages_to_device_inbox ( self , local_messages_by_user_then_device , remote_messages_by_destination ) : def add_messages_txn ( txn , now_ms , stream_id ) : self . _add_messages_to_local_device_inbox_txn ( txn , stream_id , local_messages_by_user_then_device ) sql = ( "INSERT INTO device_federation_outbox" " (destination, stream_id, queued_ts, messages_json)" " VALUES (?,?,?,?)" ) rows = [ ] for destination , edu in remote_messages_by_destination . items ( ) : edu_json = simplejson . dumps ( edu ) rows . append ( ( destination , stream_id , now_ms , edu_json ) ) txn . executemany ( sql , rows ) with self . _device_inbox_id_gen . get_next ( ) as stream_id : now_ms = self . clock . time_msec ( ) yield self . runInteraction ( "add_messages_to_device_inbox" , add_messages_txn , now_ms , stream_id , ) for user_id in local_messages_by_user_then_device . keys ( ) : self . _device_inbox_stream_cache . entity_has_changed ( user_id , stream_id ) for destination in remote_messages_by_destination . keys ( ) : self . _device_federation_outbox_stream_cache . entity_has_changed ( destination , stream_id ) defer . returnValue ( self . _device_inbox_id_gen . get_current_token ( ) )
Used to send messages from this server.
Used to send messages from this server .
"""Used to send messages from this server.

        Args:
            sender_user_id(str): The ID of the user sending these messages.
            local_messages_by_user_and_device(dict):
                Dictionary of user_id to device_id to message.
            remote_messages_by_destination(dict):
                Dictionary of destination server_name to the EDU JSON to send.
        Returns:
            A deferred stream_id that resolves when the messages have been
            inserted.
        """ # Add the local messages directly to the local inbox. # Add the remote messages to the federation outbox. # We'll send them to a remote server when we next send a # federation transaction to that destination.
A deferred stream_id that resolves when the messages have been
inserted.
A deferred stream_id that resolves when the messages have been inserted .
self
local_messages_by_user_then_device
remote_messages_by_destination
Dictionary of destination server_name to the EDU JSON to send.
Dictionary of destination server_name to the EDU JSON to send .
sender_user_id
The ID of the user sending these messages.
The ID of the user sending these messages .
local_messages_by_user_and_device
Dictionary of user_id to device_id to message.
Dictionary of user_id to device_id to message .