As soon as any plugin was loaded, every voice frame built a full CallLegContext and event (~13 us) and scheduled its own callLater (~2-7 us), even for plugins that never look at voice. - A plugin may declare `events` (the event classes it handles). The bus keeps the union and `wants()` / `wants_any()`; the voice and data bridges check it before building anything, and emit() only calls plugins that take that event. Undeclared plugins and internal handlers still get everything. - emit_deferred batches: one call_later per reactor tick, same order. - The group context's per-stream constant part (IDs, mode, proxy flag, aliases) is resolved once per stream; each event still gets its own `extra` dict. - Fix: _plugin_started / _plugin_ended grew by one entry per call forever; END now drops the START key and ended keys are capped. Per voice frame, group voice to 6 OBP + a MASTER (48 us with no plugin): any plugin 66.0 -> 59.4 us; a data-only plugin 49.4 us. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>pull/105/head
yo
4 days ago
parent
044ba970df
commit
d74ef04bfc
@ -0,0 +1,194 @@
|
||||
# ADN DMR Peer Server - tests plugin bus event interest and batching
|
||||
#
|
||||
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
|
||||
#
|
||||
###############################################################################
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
||||
###############################################################################
|
||||
|
||||
"""Plugins pay only for the events they take; deferred events cost one call_later per tick."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from adn_server.application.plugins.application import voice_bridge as vb
|
||||
from adn_server.application.plugins.application.bus import PluginBus
|
||||
from adn_server.application.plugins.application.data_bridge import DataPluginBridge
|
||||
from adn_server.application.plugins.application.voice_bridge import VoicePluginBridge
|
||||
from adn_server.application.plugins.domain.events import (
|
||||
UnitDataEnd,
|
||||
UnitDataFrame,
|
||||
UnitDataStart,
|
||||
VoiceCallEnd,
|
||||
VoiceCallFrame,
|
||||
VoiceCallStart,
|
||||
)
|
||||
from adn_server.domain import HBPF_DATA_SYNC, HBPF_VOICE
|
||||
|
||||
|
||||
class _Plugin:
|
||||
def __init__(self, name: str, events=None) -> None:
|
||||
self.name = name
|
||||
if events is not None:
|
||||
self.events = events
|
||||
self.got: list[object] = []
|
||||
|
||||
def on_event(self, event: object) -> None:
|
||||
self.got.append(event)
|
||||
|
||||
def on_shutdown(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
CONFIG = {"GLOBAL": {"SERVER_ID": 2131}, "SYSTEMS": {"SYSTEM": {"MODE": "MASTER"}}}
|
||||
|
||||
|
||||
def _voice_frame(bridge: VoicePluginBridge, stream: int = 7) -> None:
|
||||
bridge.notify_group_after_forward(
|
||||
system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5", slot=2,
|
||||
frame_type=HBPF_VOICE, dtype_vseq=1, stream_id=stream.to_bytes(4, "big"), data=b"DMRD" + bytes(51),
|
||||
pkt_time=1.0, source_is_obp=False, obp_hops=b"", obp_source_server=None, obp_ber=b"\0",
|
||||
obp_rssi=b"\0", obp_source_rptr=b"\0\0\0\0", synthetic_announcement=False, forwarded=[],
|
||||
)
|
||||
|
||||
|
||||
def _data_frame(bridge: DataPluginBridge, dtype: int = 7) -> None:
|
||||
bridge.notify_after_forward(
|
||||
route="unit", system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\x0d\xbb\xa7",
|
||||
seq=1, slot=2, frame_type=HBPF_DATA_SYNC, dtype_vseq=dtype, stream_id=b"\0\0\0\x09",
|
||||
data=b"DMRD" + bytes(51), pkt_time=1.0, source_is_obp=False, obp_hops=b"", obp_source_server=None,
|
||||
obp_ber=b"\0", obp_rssi=b"\0", obp_source_rptr=b"\0\0\0\0", forwarded=[],
|
||||
)
|
||||
|
||||
|
||||
def test_a_plugin_without_events_still_gets_everything() -> None:
|
||||
bus = PluginBus()
|
||||
legacy = _Plugin("legacy")
|
||||
bus.register_plugin(legacy)
|
||||
assert bus.wants(VoiceCallFrame) and bus.wants(UnitDataEnd)
|
||||
_voice_frame(VoicePluginBridge(bus, CONFIG))
|
||||
assert [type(e) for e in legacy.got] == [VoiceCallFrame]
|
||||
|
||||
|
||||
def test_declared_events_are_the_only_ones_delivered() -> None:
|
||||
bus = PluginBus()
|
||||
data = _Plugin("d-aprs", events=(UnitDataFrame,))
|
||||
voice = _Plugin("igate", events=(VoiceCallFrame, VoiceCallEnd))
|
||||
bus.register_plugin(data)
|
||||
bus.register_plugin(voice)
|
||||
_voice_frame(VoicePluginBridge(bus, CONFIG))
|
||||
_data_frame(DataPluginBridge(bus, CONFIG))
|
||||
assert [type(e) for e in voice.got] == [VoiceCallFrame]
|
||||
assert [type(e) for e in data.got] == [UnitDataFrame]
|
||||
|
||||
|
||||
def test_voice_frames_are_not_even_built_when_nobody_takes_them(monkeypatch) -> None:
|
||||
bus = PluginBus()
|
||||
bus.register_plugin(_Plugin("d-aprs", events=(UnitDataStart, UnitDataFrame, UnitDataEnd)))
|
||||
bridge = VoicePluginBridge(bus, CONFIG)
|
||||
|
||||
def boom(*a, **k):
|
||||
raise AssertionError("voice event built for nobody")
|
||||
|
||||
monkeypatch.setattr(bridge, "_group_ctx_base", boom)
|
||||
monkeypatch.setattr(vb, "CallLegContext", boom)
|
||||
_voice_frame(bridge)
|
||||
assert not bridge.has_subscribers()
|
||||
|
||||
|
||||
def test_unregistering_the_last_interested_plugin_stops_the_work() -> None:
|
||||
bus = PluginBus()
|
||||
igate = _Plugin("igate", events=(VoiceCallFrame,))
|
||||
bus.register_plugin(igate)
|
||||
assert bus.wants(VoiceCallFrame)
|
||||
bus.unregister_plugin(igate)
|
||||
assert not bus.wants(VoiceCallFrame) and not bus.wants_any(VoiceCallStart, UnitDataFrame)
|
||||
|
||||
|
||||
def test_an_internal_handler_takes_everything() -> None:
|
||||
bus = PluginBus()
|
||||
bus.subscribe(lambda e: None)
|
||||
assert bus.wants(VoiceCallFrame) and bus.wants(UnitDataStart)
|
||||
|
||||
|
||||
def test_deferred_events_of_a_tick_share_one_call_later_and_keep_their_order() -> None:
|
||||
scheduled: list = []
|
||||
bus = PluginBus()
|
||||
bus.set_call_later(lambda delay, fn, *a: scheduled.append((fn, a)))
|
||||
plugin = _Plugin("p")
|
||||
bus.register_plugin(plugin)
|
||||
events = [object(), object(), object()]
|
||||
for e in events:
|
||||
bus.emit_deferred(e)
|
||||
assert len(scheduled) == 1 and plugin.got == []
|
||||
fn, args = scheduled.pop()
|
||||
fn(*args)
|
||||
assert plugin.got == events
|
||||
bus.emit_deferred(events[0]) # next tick schedules again
|
||||
assert len(scheduled) == 1
|
||||
|
||||
|
||||
def test_a_deferred_event_nobody_takes_is_not_scheduled() -> None:
|
||||
scheduled: list = []
|
||||
bus = PluginBus()
|
||||
bus.set_call_later(lambda delay, fn, *a: scheduled.append(fn))
|
||||
bus.register_plugin(_Plugin("d-aprs", events=(UnitDataFrame,)))
|
||||
bus.emit_deferred(VoiceCallFrame(context=None, dmrpkt=b"", frame_type=0, dtype_vseq=0))
|
||||
assert scheduled == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("calls", [3, vb._ENDED_KEEP + 10])
|
||||
def test_stream_bookkeeping_does_not_grow_without_bound(calls: int) -> None:
|
||||
bus = PluginBus()
|
||||
bus.register_plugin(_Plugin("p"))
|
||||
bridge = VoicePluginBridge(bus, CONFIG)
|
||||
kw = dict(system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5", slot=2,
|
||||
pkt_time=1.0, source_is_obp=False)
|
||||
for n in range(calls):
|
||||
sid = n.to_bytes(4, "big")
|
||||
assert bridge.emit_group_voice_start(stream_id=sid, **kw)
|
||||
assert bridge.emit_group_voice_end(stream_id=sid, duration_s=1.0, **kw)
|
||||
assert not bridge.emit_group_voice_end(stream_id=sid, duration_s=1.0, **kw) # still once
|
||||
assert bridge._plugin_started == set()
|
||||
assert len(bridge._plugin_ended) == min(calls, vb._ENDED_KEEP)
|
||||
|
||||
|
||||
def test_a_stream_resolves_its_aliases_once_and_each_event_owns_its_extra(monkeypatch) -> None:
|
||||
calls: list[int] = []
|
||||
real = vb.alias_extra
|
||||
monkeypatch.setattr(vb, "alias_extra", lambda *a: calls.append(1) or real(*a))
|
||||
bus = PluginBus()
|
||||
plugin = _Plugin("igate", events=(VoiceCallFrame,))
|
||||
bus.register_plugin(plugin)
|
||||
bridge = VoicePluginBridge(bus, {**CONFIG, "_SUB_IDS": {1: "C31AG"}})
|
||||
for _ in range(6):
|
||||
_voice_frame(bridge)
|
||||
assert len(calls) == 1
|
||||
assert plugin.got[0].context.extra == {"src_callsign": "C31AG"}
|
||||
plugin.got[0].context.extra["mutated"] = True
|
||||
assert "mutated" not in plugin.got[1].context.extra
|
||||
|
||||
|
||||
def test_the_cached_context_is_dropped_when_the_stream_ends() -> None:
|
||||
bus = PluginBus()
|
||||
bus.register_plugin(_Plugin("p"))
|
||||
bridge = VoicePluginBridge(bus, CONFIG)
|
||||
_voice_frame(bridge, stream=7)
|
||||
assert bridge._stream_ctx
|
||||
bridge.emit_group_voice_end(system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5",
|
||||
slot=2, stream_id=(7).to_bytes(4, "big"), pkt_time=2.0, duration_s=1.0,
|
||||
source_is_obp=False)
|
||||
assert bridge._stream_ctx == {}
|
||||
Loading…
Reference in new issue