codekingpro/portable-devtools

1"""2Vendored file: Server Side Events (SSE) client for Python.3 4Source:5- Author: Maxime Petazzoni <maxime.petazzoni@bulix.org>6- Repository: https://github.com/mpetazzoni/sseclient7- File: https://github.com/mpetazzoni/sseclient/blob/main/sseclient/__init__.py8 9License:10- Apache-2.0 (from upstream project)11 12Provides a generator of SSE received through an existing HTTP response.13"""14 15import logging16 17__author__ = 'Maxime Petazzoni <maxime.petazzoni@bulix.org>'18__email__ = 'maxime.petazzoni@bulix.org'19__all__ = ['SSEClient']20 21_FIELD_SEPARATOR = ':'22 23 24class SSEClient:25    """Implementation of a SSE client.26 27    See http://www.w3.org/TR/2009/WD-eventsource-20091029/ for the28    specification.29    """30 31    def __init__(self, event_source, char_enc='utf-8'):32        """Initialize the SSE client over an existing, ready to consume33        event source.34 35        The event source is expected to be a binary stream and have a close()36        method. That would usually be something that implements37        io.BinaryIOBase, like an httplib or urllib3 HTTPResponse object.38        """39        self._logger = logging.getLogger(self.__class__.__module__)40        self._logger.debug('Initialized SSE client from event source %s',41                           event_source)42        self._event_source = event_source43        self._char_enc = char_enc44 45    def _read(self):46        """Read the incoming event source stream and yield event chunks.47 48        Unfortunately it is possible for some servers to decide to break an49        event into multiple HTTP chunks in the response. It is thus necessary50        to correctly stitch together consecutive response chunks and find the51        SSE delimiter (empty new line) to yield full, correct event chunks."""52        data = b''53        for chunk in self._event_source:54            for line in chunk.splitlines(True):55                data += line56                if data.endswith((b'\r\r', b'\n\n', b'\r\n\r\n')):57                    yield data58                    data = b''59        if data:60            yield data61 62    def events(self):63        for chunk in self._read():64            event = Event()65            # Split before decoding so splitlines() only uses \r and \n66            for line in chunk.splitlines():67                # Decode the line.68                line = line.decode(self._char_enc)69 70                # Lines starting with a separator are comments and are to be71                # ignored.72                if not line.strip() or line.startswith(_FIELD_SEPARATOR):73                    continue74 75                data = line.split(_FIELD_SEPARATOR, 1)76                field = data[0]77 78                # Ignore unknown fields.79                if field not in event.__dict__:80                    self._logger.debug('Saw invalid field %s while parsing '81                                       'Server Side Event', field)82                    continue83 84                if len(data) > 1:85                    # From the spec:86                    # "If value starts with a single U+0020 SPACE character,87                    # remove it from value."88                    if data[1].startswith(' '):89                        value = data[1][1:]90                    else:91                        value = data[1]92                else:93                    # If no value is present after the separator,94                    # assume an empty value.95                    value = ''96 97                # The data field may come over multiple lines and their values98                # are concatenated with each other.99                if field == 'data':100                    event.__dict__[field] += value + '\n'101                else:102                    event.__dict__[field] = value103 104            # Events with no data are not dispatched.105            if not event.data:106                continue107 108            # If the data field ends with a newline, remove it.109            if event.data.endswith('\n'):110                event.data = event.data[0:-1]111 112            # Empty event names default to 'message'113            event.event = event.event or 'message'114 115            # Dispatch the event116            self._logger.debug('Dispatching %s...', event)117            yield event118 119    def close(self):120        """Manually close the event source stream."""121        self._event_source.close()122 123 124class Event:125    """Representation of an event from the event stream."""126 127    def __init__(self, id=None, event='message', data='', retry=None):128        self.id = id129        self.event = event130        self.data = data131        self.retry = retry132 133    def __str__(self):134        s = f'{self.event} event'135        if self.id:136            s += f' #{self.id}'137        if self.data:138            s += ', {} byte{}'.format(len(self.data),139                                        's' if len(self.data) else '')140        else:141            s += ', no data'142        if self.retry:143            s += f', retry in {self.retry}ms'144        return s145