from enum import Enum, auto
from getpass import getuser
from threading import Thread
-from typing import (Any, Callable, Collection, Generic, Iterator, NamedTuple,
- Optional, Self, TypeVar)
+from typing import (Any, Callable, Collection, Generic, Iterable, Iterator,
+ NamedTuple, Optional, Self, Set, TypeVar)
from uuid import uuid4
# ourselves
from ircplom.events import (
yield from self._completed
+class _CompletableStringsSet(_CompletableStringsCollection, Set):
+ _collected: set[str]
+ _completed: Optional[set[str]] = None
+
+ def __init__(self) -> None:
+ self._collected = set()
+
+ def _copy_collected(self) -> set[str]:
+ return set(self._collected)
+
+ def add(self, item: str) -> None:
+ 'Put item into collection.'
+ assert self._completed is None
+ assert item not in self._collected
+ self._collected.add(item)
+
+ def intersection(self, *others: Iterable[str]) -> set[str]:
+ 'Compare self to other set(s).'
+ assert self._completed is not None
+ result = self._completed
+ for other in others:
+ assert isinstance(other, set)
+ result = result.intersection(other)
+ return result
+
+
class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection):
_collected: tuple[str, ...] = tuple()
_completed: Optional[tuple[str, ...]] = None
self._completed = None
self._collected = tuple()
- def into_set(self) -> set[str]:
- 'Return as mere set.'
- assert self._completed is not None
- return set(self._completed)
-
class IntoUpdateValueMixin(AutoAttrMixin):
'Provides .into_update_value(), exportable for module-external consumers.'
def clear(self) -> None:
self._dict.clear()
- self._ls = _CompletableStringsOrdered()
- self._list = _CompletableStringsOrdered()
+ self._ls = _CompletableStringsSet()
+ self._list = _CompletableStringsSet()
self._list_expectations: dict[str, set[str]] = {
'ACK': set(), 'NAK': set()}
if verb in {'LS', 'LIST'}:
target = getattr(self, f'_{verb.lower()}')
for item in items:
- target.append(item)
+ target.add(item)
if complete:
target.complete()
- if target == self._ls:
+ if target is self._ls:
for cap_name in _NAMES_DESIRED_SERVER_CAPS:
self._send('REQ', cap_name)
self._send('LIST')
- elif target == self._list:
+ else:
acks = self._list_expectations['ACK']
naks = self._list_expectations['NAK']
- list_set = target.into_set()
- assert acks == list_set & acks
- assert set() == list_set & naks
+ assert acks == self._list.intersection(acks)
+ assert set() == self._list.intersection(naks)
for key, data in [_Dict.key_val_from_eq_str(entry)
for entry in self._ls]:
self._dict[key].data = data