|
| 1 | +import urllib3 |
| 2 | + |
| 3 | +from ldclient.config import HTTPConfig |
| 4 | +from ldclient.impl.http import HTTPFactory |
| 5 | +from ldclient.util import throw_if_unsuccessful_response |
| 6 | + |
| 7 | + |
| 8 | +class _BufferedLineReader: |
| 9 | + """ |
| 10 | + Helper class that encapsulates the logic for reading UTF-8 stream data as a series of text lines, |
| 11 | + each of which can be terminated by \n, \r, or \r\n. |
| 12 | + """ |
| 13 | + def lines_from(chunks): |
| 14 | + """ |
| 15 | + Takes an iterable series of encoded chunks (each of "bytes" type) and parses it into an iterable |
| 16 | + series of strings, each of which is one line of text. The line does not include the terminator. |
| 17 | + """ |
| 18 | + last_char_was_cr = False |
| 19 | + partial_line = None |
| 20 | + |
| 21 | + for chunk in chunks: |
| 22 | + if len(chunk) == 0: |
| 23 | + continue |
| 24 | + |
| 25 | + # bytes.splitlines() will correctly break lines at \n, \r, or \r\n, and is faster than |
| 26 | + # iterating through the characters in Python code. However, we have to adjust the results |
| 27 | + # in several ways as described below. |
| 28 | + lines = chunk.splitlines() |
| 29 | + if last_char_was_cr: |
| 30 | + last_char_was_cr = False |
| 31 | + if chunk[0] == 10: |
| 32 | + # If the last character we saw was \r, and then the first character in buf is \n, then |
| 33 | + # that's just a single \r\n terminator, so we should remove the extra blank line that |
| 34 | + # splitlines added for that first \n. |
| 35 | + lines.pop(0) |
| 36 | + if len(lines) == 0: |
| 37 | + continue # ran out of data, continue to get next chunk |
| 38 | + if partial_line is not None: |
| 39 | + # On our last time through the loop, we ended up with an unterminated line, so we should |
| 40 | + # treat our first parsed line here as a continuation of that. |
| 41 | + lines[0] = partial_line + lines[0] |
| 42 | + partial_line = None |
| 43 | + # Check whether the buffer really ended in a terminator. If it did not, then the last line in |
| 44 | + # lines is a partial line and should not be emitted yet. |
| 45 | + last_char = chunk[len(chunk)-1] |
| 46 | + if last_char == 13: |
| 47 | + last_char_was_cr = True # remember this in case the next chunk starts with \n |
| 48 | + elif last_char != 10: |
| 49 | + partial_line = lines.pop() # remove last element which is the partial line |
| 50 | + for line in lines: |
| 51 | + yield line.decode() |
| 52 | + |
| 53 | + |
| 54 | +class Event: |
| 55 | + """ |
| 56 | + An event received by SSEClient. |
| 57 | + """ |
| 58 | + def __init__(self, event='message', data='', last_event_id=None): |
| 59 | + self._event = event |
| 60 | + self._data = data |
| 61 | + self._id = last_event_id |
| 62 | + |
| 63 | + @property |
| 64 | + def event(self): |
| 65 | + """ |
| 66 | + The event type, or "message" if not specified. |
| 67 | + """ |
| 68 | + return self._event |
| 69 | + |
| 70 | + @property |
| 71 | + def data(self): |
| 72 | + """ |
| 73 | + The event data. |
| 74 | + """ |
| 75 | + return self._data |
| 76 | + |
| 77 | + @property |
| 78 | + def last_event_id(self): |
| 79 | + """ |
| 80 | + The last non-empty "id" value received from this stream so far. |
| 81 | + """ |
| 82 | + return self._id |
| 83 | + |
| 84 | + def dump(self): |
| 85 | + lines = [] |
| 86 | + if self.id: |
| 87 | + lines.append('id: %s' % self.id) |
| 88 | + |
| 89 | + # Only include an event line if it's not the default already. |
| 90 | + if self.event != 'message': |
| 91 | + lines.append('event: %s' % self.event) |
| 92 | + |
| 93 | + lines.extend('data: %s' % d for d in self.data.split('\n')) |
| 94 | + return '\n'.join(lines) + '\n\n' |
| 95 | + |
| 96 | + |
| 97 | +class SSEClient: |
| 98 | + """ |
| 99 | + A simple Server-Sent Events client. |
| 100 | +
|
| 101 | + This implementation does not include automatic retrying of a dropped connection; the caller will do that. |
| 102 | + If a connection ends, the events iterator will simply end. |
| 103 | + """ |
| 104 | + def __init__(self, url, last_id=None, http_factory=None, **kwargs): |
| 105 | + self.url = url |
| 106 | + self.last_id = last_id |
| 107 | + self._chunk_size = 10000 |
| 108 | + |
| 109 | + if http_factory is None: |
| 110 | + http_factory = HTTPFactory({}, HTTPConfig()) |
| 111 | + self._timeout = http_factory.timeout |
| 112 | + base_headers = http_factory.base_headers |
| 113 | + |
| 114 | + self.http = http_factory.create_pool_manager(1, url) |
| 115 | + |
| 116 | + # Any extra kwargs will be fed into the request call later. |
| 117 | + self.requests_kwargs = kwargs |
| 118 | + |
| 119 | + # The SSE spec requires making requests with Cache-Control: nocache |
| 120 | + if 'headers' not in self.requests_kwargs: |
| 121 | + self.requests_kwargs['headers'] = {} |
| 122 | + |
| 123 | + self.requests_kwargs['headers'].update(base_headers) |
| 124 | + |
| 125 | + self.requests_kwargs['headers']['Cache-Control'] = 'no-cache' |
| 126 | + |
| 127 | + # The 'Accept' header is not required, but explicit > implicit |
| 128 | + self.requests_kwargs['headers']['Accept'] = 'text/event-stream' |
| 129 | + |
| 130 | + self._connect() |
| 131 | + |
| 132 | + def _connect(self): |
| 133 | + if self.last_id: |
| 134 | + self.requests_kwargs['headers']['Last-Event-ID'] = self.last_id |
| 135 | + |
| 136 | + # Use session if set. Otherwise fall back to requests module. |
| 137 | + self.resp = self.http.request( |
| 138 | + 'GET', |
| 139 | + self.url, |
| 140 | + timeout=self._timeout, |
| 141 | + preload_content=False, |
| 142 | + retries=0, # caller is responsible for implementing appropriate retry semantics, e.g. backoff |
| 143 | + **self.requests_kwargs) |
| 144 | + |
| 145 | + # Raw readlines doesn't work because we may be missing newline characters until the next chunk |
| 146 | + # For some reason, we also need to specify a chunk size because stream=True doesn't seem to guarantee |
| 147 | + # that we get the newlines in a timeline manner |
| 148 | + self.resp_file = self.resp.stream(amt=self._chunk_size) |
| 149 | + |
| 150 | + # TODO: Ensure we're handling redirects. Might also stick the 'origin' |
| 151 | + # attribute on Events like the Javascript spec requires. |
| 152 | + throw_if_unsuccessful_response(self.resp) |
| 153 | + |
| 154 | + @property |
| 155 | + def events(self): |
| 156 | + """ |
| 157 | + An iterable series of Event objects received from the stream. |
| 158 | + """ |
| 159 | + event_type = "" |
| 160 | + event_data = None |
| 161 | + for line in _BufferedLineReader.lines_from(self.resp_file): |
| 162 | + if line == "": |
| 163 | + if event_data is not None: |
| 164 | + yield Event("message" if event_type == "" else event_type, event_data, self.last_id) |
| 165 | + event_type = "" |
| 166 | + event_data = None |
| 167 | + continue |
| 168 | + colon_pos = line.find(':') |
| 169 | + if colon_pos < 0: |
| 170 | + continue # malformed line - ignore |
| 171 | + if colon_pos == 0: |
| 172 | + continue # comment - currently we're not surfacing these |
| 173 | + name = line[0:colon_pos] |
| 174 | + if colon_pos < (len(line) - 1) and line[colon_pos + 1] == ' ': |
| 175 | + colon_pos += 1 |
| 176 | + value = line[colon_pos+1:] |
| 177 | + if name == 'event': |
| 178 | + event_type = value |
| 179 | + elif name == 'data': |
| 180 | + event_data = value if event_data is None else (event_data + "\n" + value) |
| 181 | + elif name == 'id': |
| 182 | + self.last_id = value |
| 183 | + elif name == 'retry': |
| 184 | + pass # auto-reconnect is not implemented in this simplified client |
| 185 | + # unknown field names are ignored in SSE |
| 186 | + |
| 187 | + def __enter__(self): |
| 188 | + return self |
| 189 | + |
| 190 | + def __exit__(self, type, value, traceback): |
| 191 | + self.close() |
0 commit comments