Return to os.captureEvent to prevent event losses

Author: neumond

README.md

@@ -59,15 +59,14 @@ with fs.open('filename', 'r') as f:
         print(line)
 ```
 
-Waiting for event:
+Waiting for event (`os.captureEvent` instead `os.pullEvent`):
 
 ```python
 from cc import os
 
 timer_id = os.startTimer(2)
-while True:
-    e = os.pullEvent('timer')
-    if e[1] == timer_id:
+for e in os.captureEvent('timer'):
+    if e[0] == timer_id:
         print('Timer reached')
         break
 ```

computercraft/server.py

@@ -49,7 +49,7 @@ async def ws(self, request):
         if sess is not None:
             async for msg in self._json_messages(ws):
                 if msg['action'] == 'event':
-                    pass
+                    sess.on_event(msg['event'], msg['params'])
                 elif msg['action'] == 'task_result':
                     sess.on_task_result(msg['task_id'], msg['result'])
                 else:

computercraft/sess.py

@@ -2,6 +2,7 @@
 import string
 import sys
 from code import InteractiveConsole
+from collections import deque
 from contextlib import contextmanager
 from functools import partial
 from importlib import import_module
@@ -138,8 +139,7 @@ def eval_lua(lua_code, immediate=False):
         'code': lua_code,
         'immediate': immediate,
     })
-    # do not uncomment this, or use sys.__stdout__.write
-    # print('{} → {}'.format(lua_code, repr(result)))
+    # debug('{} → {}'.format(lua_code, repr(result)))
     if not immediate:
         result = rproc.coro(result)
     return result
@@ -237,6 +237,56 @@ def switch(self, *args, **kwargs):
                 self._on_death()
 
 
+class CCEventRouter:
+    def __init__(self, on_first_sub, on_last_unsub, resume_task):
+        self._stacks = {}
+        self._active = {}
+        self._on_first_sub = on_first_sub
+        self._on_last_unsub = on_last_unsub
+        self._resume_task = resume_task
+
+    def sub(self, task_id, event):
+        if event not in self._stacks:
+            self._stacks[event] = {}
+            self._on_first_sub(event)
+        se = self._stacks[event]
+        if task_id in se:
+            raise Exception('Same task subscribes to the same event twice')
+        se[task_id] = deque()
+
+    def unsub(self, task_id, event):
+        if event not in self._stacks:
+            return
+        self._stacks[event].pop(task_id, None)
+        if len(self._stacks[event]) == 0:
+            self._on_last_unsub(event)
+            del self._stacks[event]
+
+    def on_event(self, event, params):
+        if event not in self._stacks:
+            self._on_last_unsub(event)
+            return
+        for task_id, queue in self._stacks[event].items():
+            queue.append(params)
+            if self._active.get(task_id) == event:
+                self._set_task_status(task_id, event, False)
+                self._resume_task(task_id)
+
+    def get_from_stack(self, task_id, event):
+        queue = self._stacks[event][task_id]
+        try:
+            return queue.popleft()
+        except IndexError:
+            self._set_task_status(task_id, event, True)
+            return None
+
+    def _set_task_status(self, task_id, event, waits: bool):
+        if waits:
+            self._active[task_id] = event
+        else:
+            self._active.pop(task_id, None)
+
+
 class CCSession:
     def __init__(self, computer_id, sender):
         # computer_id is unique identifier of a CCSession
@@ -246,6 +296,11 @@ def __init__(self, computer_id, sender):
         self._greenlets = {}
         self._server_greenlet = get_current_greenlet()
         self._program_greenlet = None
+        self._evr = CCEventRouter(
+            lambda event: self._sender({'action': 'sub', 'event': event}),
+            lambda event: self._sender({'action': 'unsub', 'event': event}),
+            lambda task_id: self._greenlets[task_id].defer_switch('event'),
+        )
 
     def on_task_result(self, task_id, result):
         assert get_current_greenlet() is self._server_greenlet
@@ -254,6 +309,9 @@ def on_task_result(self, task_id, result):
             return
         self._greenlets[task_id].switch(result)
 
+    def on_event(self, event, params):
+        self._evr.on_event(event, params)
+
     def create_task_id(self):
         return next(self._tid_allocator)
 

computercraft/subapis/os.py

@@ -1,8 +1,8 @@
 from typing import Optional, List
 
 from ..lua import LuaNum
-from ..rproc import nil, string, option_string, number, integer, boolean, any_list
-from ..sess import eval_lua_method_factory
+from ..rproc import nil, string, option_string, number, integer, boolean
+from ..sess import eval_lua_method_factory, get_current_greenlet
 
 
 method = eval_lua_method_factory('os.')
@@ -14,8 +14,7 @@
     'getComputerLabel',
     'setComputerLabel',
     'run',
-    'pullEvent',
-    'pullEventRaw',
+    'captureEvent',
     'queueEvent',
     'clock',
     'time',
@@ -51,12 +50,21 @@ def run(environment: dict, programPath: str, *args: List[str]):
     return boolean(method('run', environment, programPath, *args))
 
 
-def pullEvent(event: str = None) -> tuple:
-    return tuple(any_list(method('pullEvent', event)))
-
-
-def pullEventRaw(event: str = None) -> tuple:
-    return tuple(any_list(method('pullEventRaw', event)))
+def captureEvent(event: str):
+    glet = get_current_greenlet().cc_greenlet
+    sess = glet._sess
+    evr = sess._evr
+    evr.sub(glet._task_id, event)
+    try:
+        while True:
+            val = evr.get_from_stack(glet._task_id, event)
+            if val is None:
+                res = sess._server_greenlet.switch()
+                assert res == 'event'
+            else:
+                yield val
+    finally:
+        evr.unsub(glet._task_id, event)
 
 
 def queueEvent(event: str, *params):

computercraft/subapis/peripheral.py

@@ -128,21 +128,19 @@ def _side(self):
         return self._prepend_params[0]
 
     def receive(self, channel: int):
-        from .os import pullEvent
+        from .os import captureEvent
 
         if self.isOpen(channel):
             raise Exception('Channel is busy')
 
         self.open(channel)
         try:
-            while True:
-                evt = pullEvent('modem_message')
-                assert evt[0] == 'modem_message'
-                if evt[1] != self._side:
+            for evt in captureEvent('modem_message'):
+                if evt[0] != self._side:
                     continue
-                if evt[2] != channel:
+                if evt[1] != channel:
                     continue
-                yield ModemMessage(*evt[3:])
+                yield ModemMessage(*evt[2:])
         finally:
             self.close(channel)
 

examples/test_os.py

@@ -21,9 +21,8 @@
 
 with _lib.assert_takes_time(1.5, 3):
     timer_id = os.startTimer(2)
-    while True:
-        e = os.pullEvent('timer')
-        if e[1] == timer_id:
+    for e in os.captureEvent('timer'):
+        if e[0] == timer_id:
             print('Timer reached')
             break
 

examples/timer.py

@@ -0,0 +1,8 @@
+from cc import os
+
+
+timer_id = os.startTimer(2)
+for e in os.captureEvent('timer'):
+    if e[0] == timer_id:
+        print('Timer reached')
+        break