diff --git a/src/zeroconf/_services/info.py b/src/zeroconf/_services/info.py index b75d6277..02b7137a 100644 --- a/src/zeroconf/_services/info.py +++ b/src/zeroconf/_services/info.py @@ -172,7 +172,7 @@ def __init__( self.priority = priority self.server = server if server else None self.server_key = server.lower() if server else None - self._properties: Dict[Union[str, bytes], Optional[Union[str, bytes]]] = {} + self._properties: Optional[Dict[Union[str, bytes], Optional[Union[str, bytes]]]] = None if isinstance(properties, bytes): self._set_text(properties) else: @@ -226,7 +226,7 @@ def addresses(self, value: List[bytes]) -> None: self._ipv6_addresses.append(addr) @property - def properties(self) -> Dict: + def properties(self) -> Dict[Union[str, bytes], Optional[Union[str, bytes]]]: """If properties were set in the constructor this property returns the original dictionary of type `Dict[Union[bytes, str], Any]`. @@ -234,6 +234,10 @@ def properties(self) -> Dict: bytes and the values are either bytes, if there was a value, even empty, or `None`, if there was none. No further decoding is attempted. The type returned is `Dict[bytes, Optional[bytes]]`. """ + if self._properties is None: + self._unpack_text_into_properties() + if TYPE_CHECKING: + assert self._properties is not None return self._properties async def async_wait(self, timeout: float) -> None: @@ -317,10 +321,10 @@ def parsed_scoped_addresses(self, version: IPVersion = IPVersion.All) -> List[st for addr in self._ip_addresses_by_version_value(version.value) ] - def _set_properties(self, properties: Dict) -> None: + def _set_properties(self, properties: Dict[Union[str, bytes], Optional[Union[str, bytes]]]) -> None: """Sets properties and text of this info from a dictionary""" self._properties = properties - list_ = [] + list_: List[bytes] = [] result = b'' for key, value in properties.items(): if isinstance(key, str): @@ -338,14 +342,25 @@ def _set_properties(self, properties: Dict) -> None: def _set_text(self, text: bytes) -> None: """Sets properties and text given a text field""" + if text == self.text: + return self.text = text + # Clear the properties cache + self._properties = None + + def _unpack_text_into_properties(self) -> None: + """Unpacks the text field into properties""" + text = self.text end = len(text) if end == 0: + # Properties should be set atomically + # in case another thread is reading them self._properties = {} return + result: Dict[Union[str, bytes], Optional[Union[str, bytes]]] = {} index = 0 - strs = [] + strs: List[bytes] = [] while index < end: length = text[index] index += 1 @@ -355,17 +370,20 @@ def _set_text(self, text: bytes) -> None: key: bytes value: Optional[bytes] for s in strs: - try: - key, value = s.split(b'=', 1) - except ValueError: + key_value = s.split(b'=', 1) + if len(key_value) == 2: + key, value = key_value + else: # No equals sign at all key = s value = None # Only update non-existent properties - if key and result.get(key) is None: + if key and key not in result: result[key] = value + # Properties should be set atomically + # in case another thread is reading them self._properties = result def get_name(self) -> str: