Transactions.

src/snek/app.py

@@ -61,7 +61,7 @@ class Application(BaseApplication):
             middlewares=middlewares, template_path=self.template_path, *args, **kwargs
         )
         session_setup(self, EncryptedCookieStorage(SESSION_KEY))
-        self.tasks = []
+        self.tasks = asyncio.Queue()
         self._middlewares.append(session_middleware)
         self.jinja2_env.add_extension(MarkdownExtension)
         self.jinja2_env.add_extension(LinkifyExtension)
@@ -76,19 +76,14 @@ class Application(BaseApplication):
         self.on_startup.append(self.prepare_database)
 
     async def create_task(self, task):
-        self.tasks.append(task)
+        await self.tasks.put(task)
 
     async def task_runner(self):
         while True:
-            await asyncio.sleep(0.1)
-            task = None
-            try:
-                task = self.tasks.pop(0)
-            except IndexError:
-                continue 
+            task = await self.tasks.get() 
             try:
                 await task
-            except:
+            except Exception as ex:
                 print(ex)
 
     async def prepare_database(self,app):

src/snek/service/socket.py

@@ -62,12 +62,12 @@ class SocketService(BaseService):
     async def broadcast(self, channel_uid, message):
         count = 0
         async for channel_member in self.app.services.channel_member.find(channel_uid=channel_uid):
-            count += await self.send_to_user(channel_member["user_uid"],message)
-        return count
+            await self.send_to_user(channel_member["user_uid"],message)
+        return True 
     
     async def delete(self, ws):
         for s in [sock for sock in self.sockets if sock.ws == ws]:
             await s.close()
             self.sockets.remove(s)
             
-    
+