Use _write_async_message for sensor updates #100
Conversation
Clients that don't keep up with sensor updates will now be disconnected, instead of causing the server to use an ever-increasing amount of RAM. This required some unintuitive changes to keep mypy happy, due to python/mypy#17723.
| # list() is to avoid errors if the set is modified during iteration. | ||
| for classic_observer in list(self._classic_observers): | ||
| classic_observer(self, reading) | ||
| for delta_observer in self._delta_observers: | ||
| for delta_observer in list(self._delta_observers): |
There was a problem hiding this comment.
Noted. Although, how does it do that? By wrapping in list(..), does it e.g. 'keep the cast copy in place until we're done using it'?
There was a problem hiding this comment.
list makes a copy. So we're iterating over a copy (which won't change) rather than the original (which might).
| @@ -99,7 +101,7 @@ def sensor_update(self, s: sensor.Sensor, reading: sensor.Reading) -> None: | |||
| msg = core.Message.inform( | |||
| "sensor-status", reading.timestamp, 1, s.name, reading.status, reading.value | |||
| ) | |||
| self.write_message(msg) | |||
| self.owner._write_async_message(self, msg) | |||
There was a problem hiding this comment.
I thought this sounded familiar. I once had to server.mass_inform, I see it uses this under the hood.
| @@ -64,6 +64,8 @@ | |||
| class ClientConnection(connection.Connection): | |||
| """Server's view of the connection from a single client.""" | |||
|
|
|||
| owner: "DeviceServer" | |||
There was a problem hiding this comment.
To get a bit pedantic/academic, your reasoning for making this a class attribute rather than an instance attribute?
There was a problem hiding this comment.
It is an instance attribute. If it was a class attribute, it would be declared
owner: ClassVar["DeviceServer"]
I added this to refine the type information: from the base class, typing tools would only know that it is a _ConnectionOwner[ClientConnection].
| async def test_slow_client_sensor( | ||
| server: DummyServer, reader: asyncio.StreamReader, writer: asyncio.StreamWriter | ||
| ) -> None: | ||
| server.max_backlog = 32768 | ||
| big_str = b"x" * 200000 | ||
| writer.write(b"?sensor-sampling str-sensor auto\n") | ||
| await check_reply( | ||
| reader, | ||
| [ | ||
| re.compile(rb"^#sensor-status [0-9.]+ 1 str-sensor unknown \\@"), | ||
| b"!sensor-sampling ok str-sensor auto\n", | ||
| ], | ||
| ) | ||
| assert len(server._connections) == 1 | ||
| server.sensors["str-sensor"].value = big_str | ||
| assert len(server._connections) == 0 |
|
I'm going to merge for now, but I think I still want to tweak things so that the initial set of sensor updates that occur when issuing |
Clients that don't keep up with sensor updates will now be disconnected,
instead of causing the server to use an ever-increasing amount of RAM.
This required some unintuitive changes to keep mypy happy, due to
python/mypy#17723.