def _copy_collected(self) -> set[str]:
return set(self._collected)
- def add(self, item: str) -> None:
+ def _on_set(self, m_name: str, item: str, on_complete: bool, exists: bool
+ ) -> None:
+ method = getattr(self._collected, m_name)
+ if on_complete:
+ assert self._completed is not None
+ assert (item in self._completed) == exists
+ assert (item in self._collected) == exists
+ method(item)
+ self.complete()
+ else:
+ assert self._completed is None
+ assert (item in self._collected) == exists
+ method(item)
+
+ def completable_add(self, item: str, on_complete: bool) -> None:
'Put item into collection.'
- assert self._completed is None
- assert item not in self._collected
- self._collected.add(item)
+ self._on_set('add', item, on_complete, False)
+
+ def completable_remove(self, item: str, on_complete: bool) -> None:
+ 'Remove item from collection.'
+ self._on_set('remove', item, on_complete, True)
def intersection(self, *others: Iterable[str]) -> set[str]:
'Compare self to other set(s).'
if complete:
self.complete()
- def remove(self, value: str, complete=False) -> None:
- 'Remove value from list.'
- assert value in self._collected
- self._collected = tuple(x for x in self._collected if x != value)
- if complete:
- self.complete()
-
def clear(self) -> None:
self._completed = None
self._collected = tuple()
'Return non-updating copy of self.'
if isinstance(self, _Dict):
return None
- if isinstance(self, _CompletableStringsOrdered):
+ if isinstance(self, _CompletableStringsCollection):
return self._completed
if isinstance(self, _CompletableTopic):
return Topic(*self._completed)
self._on_update()
-class _UpdatingCompletableStringsOrdered(_UpdatingCompletable,
- _CompletableStringsOrdered):
+class _UpdatingCompletableStringsSet(
+ _UpdatingCompletable, _CompletableStringsSet):
+ pass
+
+
+class _UpdatingCompletableStringsOrdered(
+ _UpdatingCompletable, _CompletableStringsOrdered):
pass
class _Channel:
- user_ids: _CompletableStringsOrdered
+ user_ids: _CompletableStringsSet
topic: _CompletableTopic
def __init__(self,
for item in items:
n_u_h = NickUserHost(item.lstrip(self._get_membership_prefixes()))
user_id = self._userid_for_nickuserhost(n_u_h, create_if_none=True)
- self.user_ids.append(user_id, complete=False)
+ self.user_ids.completable_add(user_id, on_complete=False)
- def append_user(self, user: '_User') -> None:
- 'To .user_ids append user.nickname and declare .user_ids complete.'
+ def add_user(self, user: '_User') -> None:
+ 'To .user_ids add user.nickname, keep .user_ids declared complete.'
user_id = self._userid_for_nickuserhost(user, create_if_none=True,
updating=True)
- self.user_ids.append(user_id, complete=True)
+ self.user_ids.completable_add(user_id, on_complete=True)
def remove_user(self, user: '_User') -> None:
- 'From .user_ids remove .nickname and declare .user_ids complete.'
- self.user_ids.remove(user.id_, complete=True)
+ 'From .user_ids remove .nickname, keep .user_ids declared complete.'
+ self.user_ids.completable_remove(user.id_, on_complete=True)
self.purge_users()
class _UpdatingChannel(_UpdatingMixin, _Channel):
- user_ids: _UpdatingCompletableStringsOrdered
+ user_ids: _UpdatingCompletableStringsSet
topic: _UpdatingCompletableTopic
if verb in {'LS', 'LIST'}:
target = getattr(self, f'_{verb.lower()}')
for item in items:
- target.add(item)
+ target.completable_add(item, False)
if complete:
target.complete()
if target is self._ls:
else:
self.caps.end_negotiation()
elif ret['_verb'] == 'JOIN' and ret['joiner'] != self.db.users['me']:
- self.db.channels[ret['channel']].append_user(ret['joiner'])
+ self.db.channels[ret['channel']].add_user(ret['joiner'])
elif ret['_verb'] == 'NICK':
user_id = self.db.users.id_for_nickuserhost(ret['named'],
updating=True)