diff --git a/tests/test_input.py b/tests/test_input.py index c1bfb68..5ad5547 100644 --- a/tests/test_input.py +++ b/tests/test_input.py @@ -648,7 +648,7 @@ class PadEvents(unittest.TestCase): self.assertEqual(out[3:5], [{"button": "left", "down": True}, {"button": "left", "down": False}]) self.assertEqual(out[5], {"scroll": [0, -15]}) # up = content follows the finger self.assertEqual(out[6:], [{"button": "right", "down": True}, {"button": "right", "down": False}]) - self.assertEqual(extra, "") + self.assertEqual(extra, []) def test_keys_and_modifiers(self): out, _ = self.pad([{"specialKey": 12}, {"specialKey": 1, "ctrl": True}, {"key": "c", "ctrl": True}]) @@ -657,11 +657,101 @@ class PadEvents(unittest.TestCase): self.assertEqual([(e["key"], e["down"]) for e in out[6:]], [(29, True), (46, True), (46, False), (29, False)]) # Ctrl+C - def test_text_and_accents(self): - out, extra = self.pad([{"key": "Hi "}, {"key": "café 😀"}]) - self.assertEqual(out, [{"text": "Hi caf "}]) - self.assertEqual(extra, "é😀") + def test_text_stays_ascii_through_gamescope(self): + out, slow = self.pad([{"key": "Hi "}, {"key": "there"}]) + self.assertEqual(out, [{"text": "Hi there"}]) + self.assertEqual(slow, []) + + def test_keyboard_goes_to_kde_whole_and_in_order(self): + events = [{"key": "a"}, {"key": "é"}, {"specialKey": 12}, {"key": "c", "ctrl": True}, {"dx": 2, "dy": 1}] + out, slow = self.pad(events, kde=True) + self.assertEqual(out, [{"dx": 2, "dy": 1}]) # the pointer still goes through gamescope + self.assertEqual(slow, events[:4]) # the keyboard as sent, in order def test_shifted_key_with_modifier(self): out, _ = self.pad([{"key": "A", "ctrl": True}]) self.assertEqual([e["key"] for e in out], [29, 42, 30, 30, 42, 29]) + + +class PadDelivery(unittest.TestCase): + """remote_input's promises: "sent" means the whole batch is taken; accents go once, in order.""" + + class Agent: + def __init__(self, sent=True, state="ready"): + self.sent, self.state, self.got = sent, state, [] + + def send(self, events): + if events and self.sent: + self.got.append(events) + return {"state": self.state, "sent": self.sent and bool(events)} + + def setUp(self): + import server + self.s = server + self.saved = (server._touch, server._input, server.KDE_WAIT) + server._touch, server._input = self.Agent(), self.Agent() + server._kde_queue.clear() + + def tearDown(self): + self.s._touch, self.s._input, self.s.KDE_WAIT = self.saved + self.s._kde_queue.clear() + + def settle(self): + worker = self.s._kde_worker[0] + if worker: + worker.join(5) + + def typed(self): + return [e for batch in self.s._input.got for e in batch] + + def test_accent_only_batch_is_acknowledged_and_typed_once(self): + status = self.s.remote_input({"events": [{"key": "é"}]}) + self.assertTrue(status["sent"]) # so the page doesn't send it again + self.settle() + self.assertEqual(self.typed(), [{"key": "é"}]) + + def test_mixed_text_stays_in_one_ordered_stream(self): + self.s.remote_input({"events": [{"key": "aéb"}, {"specialKey": 12}]}) + self.settle() + self.assertEqual(self.typed(), [{"key": "aéb"}, {"specialKey": 12}]) + self.assertEqual(self.s._touch.got, []) # none of it split off through gamescope + + def test_batch_gamescope_didnt_take_is_not_queued(self): + self.s._touch = self.Agent(sent=False, state="starting") + status = self.s.remote_input({"events": [{"dx": 1, "dy": 1}, {"key": "é"}]}) + self.assertFalse(status["sent"]) # the page tries the whole batch again... + self.assertEqual(self.s._kde_queue, []) # ...so the accent isn't queued twice + + def test_waits_for_kde_connect_then_sends_without_more_input(self): + self.s._input = self.Agent(sent=False, state="starting") + self.s.remote_input({"events": [{"key": "é"}]}) + time.sleep(0.2) + self.assertEqual(self.s._kde_queue, [{"key": "é"}]) + self.s._input.sent = True + self.settle() + self.assertEqual(self.typed(), [{"key": "é"}]) + self.assertEqual(self.s._kde_queue, []) + + def test_later_ascii_waits_behind_a_pending_accent(self): + self.s._input = self.Agent(sent=False, state="starting") + self.s.remote_input({"events": [{"key": "é"}]}) + self.s.remote_input({"events": [{"key": "x"}]}) + self.assertEqual(self.s._touch.got, []) # "x" can't overtake the "é" + self.assertEqual(self.s._kde_queue, [{"key": "é"}, {"key": "x"}]) + self.s._input.sent = True + self.settle() + self.assertEqual(self.typed(), [{"key": "é"}, {"key": "x"}]) + + def test_queue_is_bounded_and_dropped_on_error(self): + self.s._input = self.Agent(sent=False, state="error") + self.s.remote_input({"events": [{"key": "é"}] * 150}) + self.s.remote_input({"events": [{"key": "é"}] * 150}) + self.settle() + self.assertEqual(self.s._kde_queue, []) + self.s._input = self.Agent(sent=False, state="starting") + self.s.KDE_WAIT = 0.3 + self.s.remote_input({"events": [{"key": "é"}] * 150}) + self.s.remote_input({"events": [{"key": "é"}] * 150}) + self.assertLessEqual(len(self.s._kde_queue), self.s.KDE_QUEUE_LIMIT) + self.settle() + self.assertEqual(self.s._kde_queue, []) diff --git a/ui/server.py b/ui/server.py index 93816ec..4c7e055 100755 --- a/ui/server.py +++ b/ui/server.py @@ -856,14 +856,23 @@ PAD_CLICKS = {"singleclick": "left", "rightclick": "right", "middleclick": "midd PAD_NOTCH = 15 # scroll units for one wheel notch -def pad_events(events): - """The trackpad's and key row's events as (gamescope events, text for KDE Connect). +def typed_in_kde(events): + """Whether these events must go the KDE Connect way: they hold an accent or emoji, or + earlier ones are still waiting there (typing must stay in order, so it can't be split).""" + with _kde_lock: + waiting = bool(_kde_queue) + return waiting or any(ch not in frame_touch.ASCII for e in events for ch in e.get("key", "")) + + +def pad_events(events, kde=False): + """The trackpad's and key row's events as (gamescope events, KDE Connect events). Gamescope's own input reaches every panel and needs nothing installed, but types only - US-keyboard characters; accents and emoji go the KDE Connect way. + US-keyboard characters. With kde, the keyboard goes to KDE Connect as it was sent, so + text stays in order; the pointer always goes through gamescope. """ ascii_keys = frame_touch.ASCII - out, extra = [], [] + out, slow = [], [] def tap(code, mods): down = [{"key": c, "down": True} for c in mods] @@ -885,12 +894,15 @@ def pad_events(events): out.append({"button": "left", "down": True}) if e.get("singlerelease"): out.append({"button": "left", "down": False}) + if kde and ("key" in e or "specialKey" in e): + slow.append({k: e[k] for k in ("key", "specialKey", "ctrl", "alt", "shift", "super") if k in e}) + continue if "specialKey" in e and e["specialKey"] in PAD_KEYS: tap(PAD_KEYS[e["specialKey"]], mods) if "key" in e: for ch in e["key"]: if ch not in ascii_keys: - extra.append(ch) + continue # only reached when kde is wrong; the caller checks typed_in_kde first elif mods: code, shifted = ascii_keys[ch] tap(code, mods + ([42] if shifted and 42 not in mods else [])) @@ -899,36 +911,63 @@ def pad_events(events): out[-1]["text"] += ch else: out.append({"text": ch}) - return out, "".join(extra) + return out, slow def remote_input(body): - """{"events": [...]} sends keyboard and pointer events; {} (or none yet) just starts the agent.""" + """{"events": [...]} sends keyboard and pointer events; {} (or none yet) just starts the agent. + + "sent" in the answer means the whole batch is taken (the page keeps it and tries again + if not), so the keyboard's share is queued only once that's so. + """ events = body.get("events", []) if not isinstance(events, list) or len(events) > INPUT_BATCH_LIMIT: raise Failure(f"events must be a list of at most {INPUT_BATCH_LIMIT}", 400) events = [input_event(e) for e in events] - touch, accents = pad_events(events) + touch, slow = pad_events(events, typed_in_kde(events)) status = _touch.send(touch) - if accents: - _accents.append(accents) - _send_accents(status.get("sent") or not touch) - return status + if touch and not status.get("sent"): + return status + if slow: + _queue_for_kde(slow) + return {**status, "sent": True} -_accents = [] # accents and emoji waiting for KDE Connect, which starts the first time they're typed +# Typing KDE Connect is to do: it starts the first time it's needed, so this waits for it. +_kde_queue = [] +_kde_lock = threading.Lock() +_kde_worker = [None] +KDE_QUEUE_LIMIT = 200 +KDE_WAIT = 90 # seconds to wait for KDE Connect before giving up on what's queued -def _send_accents(wanted): - """Type what gamescope can't through KDE Connect, starting it if needed; kept until it's up.""" - if not _accents or not wanted: - return - status = _input.send([{"key": "".join(_accents)}]) - if status.get("sent"): - _accents.clear() - elif status.get("state") == "error": - _accents.clear() # not coming; don't hold them forever - del _accents[:-50] +def _queue_for_kde(events): + with _kde_lock: + _kde_queue.extend(events) + del _kde_queue[:-KDE_QUEUE_LIMIT] + if not (_kde_worker[0] and _kde_worker[0].is_alive()): + _kde_worker[0] = threading.Thread(target=_drain_kde, daemon=True) + _kde_worker[0].start() + + +def _drain_kde(): + """The only sender: in order, dropping from the queue just what went, until it's empty.""" + deadline = time.time() + KDE_WAIT + while True: + with _kde_lock: + batch = _kde_queue[:INPUT_BATCH_LIMIT] + if not batch: + return + status = _input.send(batch) + with _kde_lock: + if status.get("sent"): + del _kde_queue[:len(batch)] + deadline = time.time() + KDE_WAIT + elif status.get("state") == "error" or time.time() > deadline: + _kde_queue.clear() # not coming; don't hold it forever + return + if not status.get("sent"): + time.sleep(0.4) # ---- touch: the headset's panels, through gamescope's own input (frame_touch.py) ----