def run(self):
        """Continuously read data from the source and attempt to parse a valid
        message from the buffer of bytes. When a message is parsed, passes it
        off to the callback if one is set.
        """
        while self.running:
            try:
                payload = self.read()
            except DataSourceError as e:
                if self.running:
                    LOG.warn("Can't read from data source -- stopping: %s", e)
                break

            try:
                self.streamer
            except MissingPayloadFormatError:
                json_chars = ['\x00']
                json_chars.extend(string.printable)
                if all((char in json_chars for char in payload)):
                    self.format = "json"
                else:
                    self.format = "protobuf"
            self.streamer.receive(payload)

            while True:
                message = self.streamer.parse_next_message()
                if message is None:
                    break

                if not self._message_valid(message):
                    self.corrupted_messages += 1
                    break

                if self.callback is not None:
                    self.callback(message)
                self._receive_command_response(message)
dataset/ETHPy150Open openxc/openxc-python/openxc/sources/base.py/BytestreamDataSource.run/original