stroblme
f4fd26e44d
The Influx client was built with no timeout, so every query and write fell through to influxdb-client's own 10 s default — invisible to a flow and unchangeable. The param is in seconds like its peers; the client counts in milliseconds, so the call sites convert. The publisher backlog was a module constant, read once at import. It is the depth at which the oldest payload is dropped and the node goes degraded, and a node that bursts wants more than one that trickles, so it moves to Params and is read where the queue is built.
154 lines
4.6 KiB
Python
154 lines
4.6 KiB
Python
"""The outbound nodes reuse one connection instead of opening one per message."""
|
|
|
|
import asyncio
|
|
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import MqttNode
|
|
from fluksio.flow.nodes.http import close_shared_client, shared_client
|
|
|
|
|
|
def test_http_senders_share_one_pooled_client():
|
|
first = shared_client()
|
|
try:
|
|
assert shared_client() is first
|
|
finally:
|
|
close_shared_client()
|
|
|
|
# Closing lets the next request build a fresh one rather than reusing a
|
|
# closed pool.
|
|
assert shared_client() is not first
|
|
close_shared_client()
|
|
|
|
|
|
def _publisher() -> MqttNode:
|
|
node = MqttNode(
|
|
requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
|
|
params={"topic": {"setpoint": "heating/setpoint"}},
|
|
)
|
|
node.assign_flow("heating", "out")
|
|
return node
|
|
|
|
|
|
def test_a_started_publisher_queues_instead_of_connecting():
|
|
"""The handler runs on a worker thread; it must not block on the broker."""
|
|
node = _publisher()
|
|
|
|
async def scenario() -> None:
|
|
node._publish_queue = asyncio.Queue(maxsize=4)
|
|
node._loop = asyncio.get_running_loop()
|
|
|
|
await asyncio.to_thread(node._publisher_handler, {}, setpoint=21.0)
|
|
# call_soon_threadsafe lands on the next loop pass.
|
|
await asyncio.sleep(0)
|
|
|
|
assert node._publish_queue.qsize() == 1
|
|
assert node._publish_queue.get_nowait() == {"setpoint": 21.0}
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_full_publish_queue_drops_the_oldest():
|
|
"""A broker that cannot keep up must not grow the queue without bound."""
|
|
node = _publisher()
|
|
health: list[tuple[str, str | None]] = []
|
|
node._on_health = lambda _n, status, detail: health.append((status, detail))
|
|
|
|
async def scenario() -> None:
|
|
queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=2)
|
|
for value in (1.0, 2.0, 3.0):
|
|
node._enqueue(queue, {"setpoint": value})
|
|
|
|
assert queue.qsize() == 2
|
|
assert queue.get_nowait() == {"setpoint": 2.0}
|
|
assert queue.get_nowait() == {"setpoint": 3.0}
|
|
assert health == [("degraded", "publish queue full")]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_string_goes_on_the_wire_bare():
|
|
"""Devices on a shared broker expect `ON`, not `"ON"`."""
|
|
|
|
class Recorder:
|
|
def __init__(self) -> None:
|
|
self.published: list[tuple[str, str]] = []
|
|
|
|
async def publish(self, topic, payload, **_):
|
|
self.published.append((topic, payload))
|
|
|
|
node = MqttNode(
|
|
requires=[
|
|
MessageSpec(name="plug", port="plug", dtype=DType.STR),
|
|
MessageSpec(name="level", port="level", dtype=DType.INT),
|
|
],
|
|
params={"topic": {"plug": "actor/plug", "level": "light/level"}},
|
|
)
|
|
node.assign_flow("house", "out")
|
|
client = Recorder()
|
|
|
|
asyncio.run(node._publish_with(client, {"plug": "ON", "level": 60}))
|
|
|
|
assert client.published == [("actor/plug", "ON"), ("light/level", "60")]
|
|
|
|
|
|
def test_the_configured_timeout_reaches_the_broker_client(monkeypatch):
|
|
"""Without one, a dead socket makes the disconnect ack wait forever."""
|
|
import aiomqtt
|
|
|
|
seen: dict = {}
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
seen.update(kwargs)
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
async def publish(self, *_, **__):
|
|
return None
|
|
|
|
monkeypatch.setattr(aiomqtt, "Client", FakeClient)
|
|
|
|
node = MqttNode(
|
|
requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
|
|
params={"topic": {"setpoint": "heating/setpoint"}, "timeout": 2.5},
|
|
)
|
|
asyncio.run(node._publish_once({"setpoint": 21.0}))
|
|
|
|
assert seen["timeout"] == 2.5
|
|
|
|
|
|
def test_the_configured_backlog_reaches_the_publish_queue(monkeypatch):
|
|
"""The depth is read when the queue is built, so it has to be per node."""
|
|
import aiomqtt
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
monkeypatch.setattr(aiomqtt, "Client", FakeClient)
|
|
|
|
node = MqttNode(
|
|
requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
|
|
params={"topic": {"setpoint": "heating/setpoint"}, "publish_queue_size": 8},
|
|
)
|
|
node.assign_flow("heating", "out")
|
|
|
|
async def scenario() -> int:
|
|
await node.start_publisher()
|
|
assert node._publish_queue is not None
|
|
size = node._publish_queue.maxsize
|
|
await node.stop_publisher()
|
|
return size
|
|
|
|
assert asyncio.run(scenario()) == 8
|