Async Utilities¶
An async utility is an object instantiation with an async function that is run persistently on the asyncio event loop. It is useful for long running tasks.
For our training, we’re going to use an async utility with a queue to send messages to logged in users.
Create a utility.py
file and put the following code in it.
from guillotina.async_util import IAsyncUtility
from guillotina.component import get_multi_adapter
from guillotina.interfaces import IResourceSerializeToJsonSummary
from guillotina.utils import get_authenticated_user_id, get_current_request
import asyncio
import orjson
import logging
logger = logging.getLogger('guillotina_chat')
class IMessageSender(IAsyncUtility):
pass
class MessageSenderUtility:
def __init__(self, settings=None, loop=None):
self._loop = loop
self._settings = {}
self._webservices = []
self._closed = False
def register_ws(self, ws, request):
ws.user_id = get_authenticated_user_id()
self._webservices.append(ws)
def unregister_ws(self, ws):
self._webservices.remove(ws)
async def send_message(self, message):
summary = await get_multi_adapter(
(message, get_current_request()),
IResourceSerializeToJsonSummary)()
await self._queue.put((message, summary))
async def finalize(self):
self._closed = True
async def initialize(self, app=None):
self._queue = asyncio.Queue()
while not self._closed:
try:
message, summary = await asyncio.wait_for(self._queue.get(), 0.2)
for user_id in message.__parent__.users:
for ws in self._webservices:
if ws.user_id == user_id:
await ws.send_str(orjson.dumps(summary))
except (RuntimeError, asyncio.CancelledError, asyncio.TimeoutError):
pass
except Exception:
logger.warning(
'Error sending message',
exc_info=True)
await asyncio.sleep(1)
Async utilities must implement an initialize
method that performs the async
task. In our case, it is creating a queue and waiting to process messages
in the queue.
We will use this to send messages to registered websockets.
Like all other configured modules, make sure this file is scanned
by the packages __init__.py
file.
Additionally, async utilities need to also be configured in __init__.py
:
app_settings = {
"load_utilities": {
"guillotina_chat.message_sender": {
"provides": "guillotina_chat.utility.IMessageSender",
"factory": "guillotina_chat.utility.MessageSenderUtility",
"settings": {}
},
}
}
Sending messages¶
We’ll need to add another event subscriber to the subscribers.py
file
in order for the utility to know to send out new messages to registered
web services.So your subscribers.py
file will now look like:
from guillotina import configure
from guillotina.component import get_utility
from guillotina.interfaces import IObjectAddedEvent, IObjectModifiedEvent, IPrincipalRoleManager
from guillotina.utils import get_authenticated_user_id, get_current_request
from guillotina_chat.content import IConversation, IMessage
from guillotina_chat.utility import IMessageSender
@configure.subscriber(for_=(IConversation, IObjectAddedEvent))
@configure.subscriber(for_=(IConversation, IObjectModifiedEvent))
async def container_added(conversation, event):
user_id = get_authenticated_user_id()
if user_id not in conversation.users:
conversation.users.append(user_id)
manager = IPrincipalRoleManager(conversation)
for user in conversation.users:
manager.assign_role_to_principal(
'guillotina_chat.ConversationParticipant', user)
@configure.subscriber(for_=(IMessage, IObjectAddedEvent))
async def message_added(message, event):
utility = get_utility(IMessageSender)
await utility.send_message(message)