squash this most likely

Author: dr7ana

contrib/uwsgi-sogs-standalone.ini

@@ -27,7 +27,7 @@ processes = 2
 enable-threads = true
 http = :80
 mount = /=sogs.web:app
-mule = @(call://sogs.mule.Mule.Mule)
+mule = @(call://sogs.mule.Mule)
 log-4xx = true
 log-5xx = true
 disable-logging = true

sogs/model/bothandler.py renamed to sogs/model/manager.py

@@ -7,12 +7,15 @@
 from .. import utils
 from ..omq import send_mule
 from .user import User
+from .bot import Bot
 from .room import Room
 from .message import Message
 from .filter import SimpleFilter
 from .exc import InvalidData
+import heapq
 
-from typing import Optional, List, Union
+from dataclasses import dataclass, field
+from typing import Optional, List, Union, Any
 import time
 
 """
@@ -23,10 +26,8 @@
         - what does the bot do with the message they tried to send?
             - can store locally
             - user sends reply
-            - bot inserts it into room
-"""
+            - bot inserts it into room (?)
 
-"""
 Control Flow:
     1) message comes in HTTP request
     2) unpacked/parsed/verified/permissions checked
@@ -38,12 +39,44 @@
 """
 
 
-class BotHandler:
+@dataclass(order=True)
+class PriorityTuple(tuple):
+    priority: int
+    item: Any = field(compare=False)
+
+
+# Simple "priority queue" of bots implemented using a dict with heap
+# invariance maintained by qheap algorithm
+# TODO: when bots are designed basically, add methods for polling them
+#   and receiving their judgments
+class BotQueue:
+    def __init__(self) -> None:
+        self.queue = {}
+
+    def _qsize(self) -> int:
+        return len(self.queue.keys())
+
+    def _empty(self) -> bool:
+        return not self._qsize()
+
+    def _peek(self, priority: int):
+        return self.queue.get(priority)
+
+    def _put(self, item: PriorityTuple):
+        temp = list(self.queue.items())
+        heapq.heappush(temp, item)
+        self.queue = dict(temp)
+
+    def _get(self):
+        return heapq.heappop(self.queue)
+
+
+class Manager:
     """
     Class representing an interface that manages active bots
 
     Object Properties:
-        bots - list of bots attached to room
+        queue - BotQueue object
     """
 
     def __init__(
@@ -54,8 +87,27 @@ def __init__(
         id: Optional[int] = None,
         session_id: Optional[int] = None,
     ) -> None:
-        # immutable attributes
         self.id = id
+        self.queue = BotQueue()
+
+    def qempty(self):
+        return not self.queue._empty()
+
+    def add_bot(self, bot: Bot, priority: int = None):
+        if not priority:
+            # if no priority is given, lowest priority is assigned
+            priority = self.qsize()
+        else:
+            # if priority is already taken, find next lowest
+            while self.queue.get(priority):
+                priority += 1
+        self.queue._put(PriorityTuple(priority, bot))
+
+    def remove_bot(self):
+        do_something = 3
+
+    def peek(self, priority: int):
+        return self.queue._peek(priority)
 
     def check_permission_for(
         self,
@@ -236,5 +288,13 @@ def receive_message(
                 if whisper_to:
                     msg['whisper_to'] = whisper_to.session_id
 
+        if room._bot_status():
+            add_bot_logic = 3
+            """
+                TODO: add logic for bots receiving message and doing
+                bot things. The bots should be queried in terms of
+                priority,
+            """
+
         send_mule("message_posted", msg["id"])
         return msg

sogs/mule.py

@@ -9,13 +9,14 @@
 from . import cleanup
 from . import config
 from .omq import OMQ
-from .model.bothandler import BotHandler
+from .model.manager import Manager
 
 
 # This is the uwsgi "mule" that handles things not related to serving HTTP requests:
 # - it holds the oxenmq instance (with its own interface into sogs)
 # - it handles cleanup jobs (e.g. periodic deletions)
 
+
 def log_exceptions(f):
     @functools.wraps(f)
     def wrapper(*args, **kwargs):
@@ -27,8 +28,8 @@ def wrapper(*args, **kwargs):
 
     return wrapper
 
-class Mule:
 
+class Mule:
     def __init__(self):
         self.run()
 
@@ -57,18 +58,15 @@ def message_posted(self, m: oxenmq.Message):
         id = bt_deserialize(m.data()[0])
         app.logger.debug(f"FIXME: mule -- message posted stub, id={id}")
 
-
     @log_exceptions
     def messages_deleted(self, m: oxenmq.Message):
         ids = bt_deserialize(m.data()[0])
         app.logger.debug(f"FIXME: mule -- message delete stub, deleted messages: {ids}")
 
-
     @log_exceptions
     def message_edited(self, m: oxenmq.Message):
         app.logger.debug("FIXME: mule -- message edited stub")
 
-
     def setup_omq(self, omq: OMQ):
         app.logger.debug("Mule setting up omq")
         if isinstance(config.OMQ_LISTEN, list):
@@ -95,14 +93,14 @@ def setup_omq(self, omq: OMQ):
 
         ## NEW CODE FOR BOT
         handler = self._omq.add_category("handler", access_level=oxenmq.AuthLevel.admin)
-        handler.add_command("add_bot", add_bot)
-        handler.add_command("remove_bot", remove_bot)
-        handler.add_command("send_to_handler", message_to_handler)
+        handler.add_command("add_bot", omq.add_bot)
+        handler.add_command("remove_bot", omq.remove_bot)
+        handler.add_command("send_to_handler", omq.manager.receive_message)
 
         app.logger.debug("Mule starting omq")
         self._omq.start()
 
         # Connect mule to itself so that if something the mule does wants to send something to the mule
         # it will work.  (And so be careful not to recurse!)
         app.logger.debug("Mule connecting to self")
-        omq.mule_conn = omq.connect_inproc(on_success=None, on_failure=inproc_fail)
+        omq.mule_conn = omq.connect_inproc(on_success=None, on_failure=self.inproc_fail)

sogs/omq.py

@@ -5,30 +5,31 @@
 from oxenc import bt_serialize
 
 from . import crypto, config
-from .mule import Mule
 from .postfork import postfork
-from .model.bothandler import BotHandler
+from .model.manager import Manager
 
+omq_global = None
 
 class OMQ:
-
     @postfork
     def __init__(self):
         try:
             import uwsgi
         except ModuleNotFoundError:
             return
-        
-        self._omq = oxenmq.OxenMQ(privkey=crypto._privkey.encode(), pubkey=crypto.server_pubkey.encode())
+
+        self._omq = oxenmq.OxenMQ(
+            privkey=crypto._privkey.encode(), pubkey=crypto.server_pubkey.encode()
+        )
         self._omq.ephemeral_routing_id = True
 
-        self.bot_manager = BotHandler()
+        self.manager = Manager()
         self.test_suite = False
 
         if uwsgi.mule_id() != 0:
             uwsgi.opt['mule'].setup_omq(self)
             return
-        
+
         from .web import app  # Imported here to avoid circular import
 
         app.logger.debug(f"Starting oxenmq connection to mule in worker {uwsgi.worker_id()}")
@@ -39,6 +40,16 @@ def __init__(self):
 
         app.logger.debug(f"worker {uwsgi.worker_id()} connected to mule OMQ")
 
+        global omq_global
+        omq_global = self
+
+    def add_bot(self):
+        self.manager.add_bot()
+        # TODO: add omq logic
+
+    def remove_bot(self):
+        self.manager.remove_bot()
+        # TODO: add omq logic
 
     def send_mule(self, command, *args, prefix="worker."):
         """
@@ -47,42 +58,11 @@ def send_mule(self, command, *args, prefix="worker."):
 
         Any args will be bt-serialized and send as message parts.
         """
+        
         if prefix:
             command = prefix + command
 
-        if self.test_suite and omq is None:
+        if self.test_suite and self._omq is None:
             pass  # TODO: for mule call testing we may want to do something else here?
         else:
-            omq.send(mule_conn, command, *(bt_serialize(data) for data in args))
-
-
-# Postfork for workers: we start oxenmq and connect to the mule process
-@postfork
-def start_oxenmq():
-    try:
-        import uwsgi
-    except ModuleNotFoundError:
-        return
-    
-
-    global omq, mule_conn, bot_manager
-
-    bot_manager = BotHandler()
-
-    omq = make_omq()
-
-    if uwsgi.mule_id() != 0:
-        from . import mule
-
-        mule.setup_omq()
-        return
-
-    from .web import app  # Imported here to avoid circular import
-
-    app.logger.debug(f"Starting oxenmq connection to mule in worker {uwsgi.worker_id()}")
-
-    omq.start()
-    app.logger.debug("Started, connecting to mule")
-    mule_conn = omq.connect_remote(oxenmq.Address(config.OMQ_INTERNAL))
-
-    app.logger.debug(f"worker {uwsgi.worker_id()} connected to mule OMQ")
+            self._omq.send(self.mule_conn, command, *(bt_serialize(data) for data in args))

sogs/routes/messages.py

@@ -1,7 +1,7 @@
 from .. import http, utils
 from . import auth
 from model.room import Room
-from ..omq import send_mule
+from ..omq import omq_global
 
 from flask import abort, jsonify, g, Blueprint, request
 
@@ -361,15 +361,17 @@ def post_message(room: Room):
     """
     req = request.json
 
-    send_mule(command="send_to_handler", 
-              user=g.user,
-              room=room,
-              data=utils.decode_base64(req.get('data')),
-              sig=utils.decode_base64(req.get('signature')),
-              whisper_to=req.get('whisper_to'),
-              whisper_mods=bool(req.get('whisper_mods')),
-              files=[int(x) for x in req.get('files', [])],
-              prefix="handler.")
+    msg = omq_global.send_mule(
+        command="send_to_handler",
+        user=g.user,
+        room=room,
+        data=utils.decode_base64(req.get('data')),
+        sig=utils.decode_base64(req.get('signature')),
+        whisper_to=req.get('whisper_to'),
+        whisper_mods=bool(req.get('whisper_mods')),
+        files=[int(x) for x in req.get('files', [])],
+        prefix="handler.",
+    )
 
     return utils.jsonify_with_base64(msg), http.CREATED
 

sogs/utils.py

@@ -3,6 +3,8 @@
 from . import http
 
 import base64
+
+
 from flask import request, abort, Response
 import json
 from typing import Union, Tuple