class IntoEndnodeUpdatesMixin(AutoAttrMixin):
'Provides .into_endnode_updates exportable for module-external consumers.'
+ _cache: dict[tuple[str, ...], Any] = {}
- def into_endnode_updates(self, path: tuple[str, ...]
+ def into_endnode_updates(self,
+ client_id: str,
+ path: tuple[str, ...]
) -> list[tuple[tuple[str, ...], Any]]:
'Return path-value pairs for any update-worthy sub-elements.'
+
+ def update_unless_cached(path: tuple[str, ...], val: Any):
+ steps_so_far: list[str] = []
+ for step in path[:-1]:
+ cache_path = tuple(client_id) + tuple(steps_so_far)
+ if cache_path in self._cache:
+ del self._cache[cache_path]
+ steps_so_far += [step]
+ cache_path = tuple(client_id) + path
+ if cache_path not in self._cache or val != self._cache[cache_path]:
+ self._cache[cache_path] = val
+ return [(path, val)]
+ return []
+
if isinstance(self, _Completable):
- return ([(path, self._completed)] if self._completed is not None
- else [])
+ if self._completed is not None:
+ return update_unless_cached(path, self._completed)
+ return []
+
if isinstance(self, Dict) and not self._dict:
- return [(path, None)]
+ return update_unless_cached(path, None)
+
updates = []
for key, val in (self._dict.items() if isinstance(self, Dict)
else ((k, getattr(self, k))
for k in self._deep_annotations())):
p = path + (key,)
- updates += (val.into_endnode_updates(p)
- if isinstance(val, IntoEndnodeUpdatesMixin)
- else [(p, val)])
+ if isinstance(val, IntoEndnodeUpdatesMixin):
+ updates += val.into_endnode_updates(client_id, p)
+ else:
+ updates += update_unless_cached(p, val)
return updates
) -> None:
self._dict = caps_dict
self._send = lambda *args: sender('CAP', *args)
- self.clear(first_run=True)
+ self.clear()
- def clear(self, first_run=False) -> None:
- if not first_run:
- self._dict.clear()
+ def clear(self) -> None:
+ self._dict.clear()
self._ls = _CompletableStringsSet()
self._list = _CompletableStringsSet()
self._list_expectations: dict[str, set[str]] = {
class Client(ABC, ClientQueueMixin):
'Abstracts socket connection, loop over it, and handling messages from it.'
conn: Optional[IrcConnection] = None
- _caps: _CapsManager
_cls_conn: type[IrcConnection] = IrcConnection
def __init__(self, conn_setup: IrcConnSetup, **kwargs) -> None:
value = ((parent[last_step] if last_step in parent.keys()
else None) if isinstance(parent, Dict)
else getattr(parent, last_step))
- for path, value in (value.into_endnode_updates(path)
+ for path, value in (value.into_endnode_updates(self.client_id, path)
if isinstance(value, IntoEndnodeUpdatesMixin)
else [(path, value)]):
self._client_tui_trigger('update_db', update=_Update(path, value))
1,2 $ ?!?@? renames ?
1,2 $ users:me:user set to: [plom]
1,2 $ connection_state set to: [connected]
-1,2 $ caps cleared
2 > CAP LS :302
2 > USER plom 0 * :baz
2 < :bazbaz!~baz@baz.baz PART :#test
4 $ bazbaz!~baz@baz.baz parts
1,2 $ users:2 cleared
-, $ ?!?@? renames ?
2 < :bazbaz!~baz@baz.baz JOIN :#test
, $ ?!?@? renames ?
, $ ?!?@? renames bazbaz
4 $ bazbaz!~baz@baz.baz quits: Client Quit
1,2 $ users:2 cleared
1,2 $ users:3 cleared
-, $ ?!?@? renames ?
2 < :foo!~plom@baz.bar.foo PART :#test
1,2,3,4 $ foo!~plom@baz.bar.foo parts
1,2 $ users:3 cleared
1,2,3,4 $ ?!?@? renames ?
1,2 $ users:me:user set to: [plom]
1,2,3,4 $ connection_state set to: [connected]
-1,2 $ caps cleared
2 > CAP LS :302
2 > USER plom 0 * :baz
2 > NICK :foo
1,2 $ caps:bar:data set to: []
1,2 $ caps:baz:data set to: []
1,2 $ caps:foo:data set to: []
-1,2 $ caps:sasl:data set to: []
1,2 $ caps:sasl:data set to: [PLAIN,EXTERNAL]
1,2 $ caps:sasl:enabled set to: [True]
1,2 $ sasl_auth_state set to: [attempting]