(self, event_message: EventMessage, options: dict)
return event_message
Publish an event
Publish an event
Publish an event
async def send_event(self, event_message: EventMessage, options: dict):
        """Publish an event"""
        stream = self._get_stream_names(
            listen_for=[(event_message.api_name, event_message.event_name)]
        )[0]

        logger.debug(
            LBullets(
                L(
                    "Enqueuing event message {} in Redis stream {}",
                    Bold(event_message),
                    Bold(stream),
                ),
                items=dict(**event_message.get_metadata(), kwargs=event_message.get_kwargs()),
            )
        )

        # Performance: I suspect getting a connection from the connection manager each time is causing
        # performance issues. Need to confirm.
        with await self.connection_manager() as redis:
            start_time = time.time()
            native_id: bytes = await redis.xadd(
                stream=stream,
                fields=self.serializer(event_message),
                max_len=self.max_stream_length or None,
                exact_len=False,
            )

        logger.debug(
            L(
                "Enqueued event message {} in Redis in {} stream {}",
                Bold(event_message.canonical_name),
                human_time(time.time() - start_time),
                Bold(stream),
            )
        )

        event_message.native_id = native_id.decode("utf8")
        return event_message
async def send_event ( self , event_message : EventMessage , options : dict ) : stream = self . _get_stream_names ( listen_for = [ ( event_message . api_name , event_message . event_name ) ] ) [ 0 ] logger . debug ( LBullets ( L ( "Enqueuing event message {} in Redis stream {}" , Bold ( event_message ) , Bold ( stream ) , ) , items = dict ( * * event_message . get_metadata ( ) , kwargs = event_message . get_kwargs ( ) ) , ) ) # Performance: I suspect getting a connection from the connection manager each time is causing # performance issues. Need to confirm. with await self . connection_manager ( ) as redis : start_time = time . time ( ) native_id : bytes = await redis . xadd ( stream = stream , fields = self . serializer ( event_message ) , max_len = self . max_stream_length or None , exact_len = False , ) logger . debug ( L ( "Enqueued event message {} in Redis in {} stream {}" , Bold ( event_message . canonical_name ) , human_time ( time . time ( ) - start_time ) , Bold ( stream ) , ) ) event_message . native_id = native_id . decode ( "utf8" ) return event_message