Repository: btubbs/sseclient

Issue:
The first event retrieved by sseclient is always empty
`
.>>> from sseclient import SSEClient
.>>> client = SSEClient(url='https://stream.wikimedia.org/v2/stream/recentchange')
.>>> data = next(client)
.>>> data.__dict__
{'data': '', 'event': 'message', 'id': None, 'retry': None}
.>>> data = next(client)
.>>> data.__dict__
{'data': '{"$schema":"/mediawiki/recentchange/1.0.0","meta":{"uri":"https://commons.wikimedia.org/wiki/File:Geschichte_des_Schultei%C3%9Fenamts_und_der_Stadt_Neumarkt_014.jpg","request_id":"Xj6I2ApAED4AADHcxFEAAABL","id":"7573678b-87b2-4325-81f8-e338613e537b","dt":"2020-02-08T10:09:30Z","domain":"commons.wikimedia.org","stream":"mediawiki.recentchange","topic":"eqiad.mediawiki.recentchange","partition":0,"offset":2157854683},"id":1327149461,"type":"log","namespace":6,"title":"File:Geschichte des Schulteißenamts und der Stadt Neumarkt 014.jpg","comment":"User created page with UploadWizard","timestamp":1581156570,"user":"DALIBRI","bot":false,"log_id":290782528,"log_type":"upload","log_action":"upload","log_params":{"img_sha1":"5oe8esm1e76smmt6zbv811c0m7ay1jo","img_timestamp":"20200208100930"},"log_action_comment":"uploaded &quot;[[File:Geschichte des Schulteißenamts und der Stadt Neumarkt 014.jpg]]&quot;: User created page with UploadWizard","server_url":"https://commons.wikimedia.org","server_name":"commons.wikimedia.org","server_script_path":"/w","wiki":"commonswiki","parsedcomment":"User created page with UploadWizard"}', 'event': 'message', 'id': '[{"topic":"eqiad.mediawiki.recentchange","partition":0,"timestamp":1581156570001},{"topic":"codfw.mediawiki.recentchange","partition":0,"offset":-1}]', 'retry': None}
>>> 
`

Discussion:

Patch:
diff --git a/README.rst b/README.rst
index 04e66ab..d424ec5 100644
--- a/README.rst
+++ b/README.rst
@@ -7,6 +7,11 @@ streams (also known as EventSource, after the name of the Javascript interface
 inside browsers).  The SSEClient class accepts a url on init, and is then an
 iterator over messages coming from the server.
 
+Maintenance
+-----------
+
+I (btubbs) haven't been using this library in my own work for years, so I put limited time into maintaining it.  I will check in on pull requests and issues once per month.  If you are interested in providing more active support for the library, please reach out.
+
 Installation
 ------------
 
diff --git a/setup.py b/setup.py
index 217f8e2..eb5d8d4 100644
--- a/setup.py
+++ b/setup.py
@@ -13,7 +13,7 @@ pytest_runner = ['pytest_runner>=2.1'] if needs_pytest else []
 
 setup(
     name='sseclient',
-    version='0.0.26',
+    version='0.0.27',
     author='Brent Tubbs',
     author_email='brent.tubbs@gmail.com',
     py_modules=['sseclient'],
diff --git a/sseclient.py b/sseclient.py
index d2193e2..3bacccf 100644
--- a/sseclient.py
+++ b/sseclient.py
@@ -14,7 +14,7 @@ import six
 
 import requests
 
-__version__ = '0.0.26'
+__version__ = '0.0.27'
 
 # Technically, we should support streams that mix line endings.  This regex,
 # however, assumes that a system will provide consistent line endings.
@@ -55,8 +55,8 @@ class SSEClient(object):
         requester = self.session or requests
         self.resp = requester.get(self.url, stream=True, **self.requests_kwargs)
         self.resp_iterator = self.iter_content()
-        self.decoder = codecs.getincrementaldecoder(
-            self.resp.encoding)(errors='replace')
+        encoding = self.resp.encoding or self.resp.apparent_encoding
+        self.decoder = codecs.getincrementaldecoder(encoding)(errors='replace')
 
         # TODO: Ensure we're handling redirects.  Might also stick the 'origin'
         # attribute on Events like the Javascript spec requires.
@@ -67,12 +67,13 @@ class SSEClient(object):
             while True:
                 if hasattr(self.resp.raw, '_fp') and \
                         hasattr(self.resp.raw._fp, 'fp') and \
-                        hasattr(self.resp.raw._fp.fp, 'read1'):
+                        hasattr(self.resp.raw._fp.fp, 'read1') and \
+                        not self.resp.raw.chunked:
                     chunk = self.resp.raw._fp.fp.read1(self.chunk_size)
                 else:
-                    # _fp is not available, this means that we cannot use short
-                    # reads and this will block until the full chunk size is
-                    # actually read
+                    # _fp is not available or we are using chunked encoding
+                    # this means that we cannot use short reads and this will
+                    # block until the full chunk size is actually read
                     chunk = self.resp.raw.read(self.chunk_size)
                 if not chunk:
                     break
@@ -87,40 +88,47 @@ class SSEClient(object):
         return self
 
     def __next__(self):
-        while not self._event_complete():
-            try:
-                next_chunk = next(self.resp_iterator)
-                if not next_chunk:
-                    raise EOFError()
-                self.buf += self.decoder.decode(next_chunk)
-
-            except (StopIteration, requests.RequestException, EOFError, six.moves.http_client.IncompleteRead) as e:
-                print(e)
-                time.sleep(self.retry / 1000.0)
-                self._connect()
-
-                # The SSE spec only supports resuming from a whole message, so
-                # if we have half a message we should throw it out.
-                head, sep, tail = self.buf.rpartition('\n')
-                self.buf = head + sep
-                continue
-
-        # Split the complete event (up to the end_of_field) into event_string,
-        # and retain anything after the current complete event in self.buf
-        # for next time.
-        (event_string, self.buf) = re.split(end_of_field, self.buf, maxsplit=1)
-        msg = Event.parse(event_string)
-
-        # If the server requests a specific retry delay, we need to honor it.
-        if msg.retry:
-            self.retry = msg.retry
-
-        # last_id should only be set if included in the message.  It's not
-        # forgotten if a message omits it.
-        if msg.id:
-            self.last_id = msg.id
-
-        return msg
+        while True: #loop until event emitted
+            while not self._event_complete():
+                try:
+                    next_chunk = next(self.resp_iterator)
+                    if not next_chunk:
+                        raise EOFError()
+                    self.buf += self.decoder.decode(next_chunk)
+
+                except (StopIteration, requests.RequestException, EOFError, six.moves.http_client.IncompleteRead) as e:
+                    print(e)
+                    time.sleep(self.retry / 1000.0)
+                    self._connect()
+
+                    # The SSE spec only supports resuming from a whole message, so
+                    # if we have half a message we should throw it out.
+                    head, sep, tail = self.buf.rpartition('\n')
+                    self.buf = head + sep
+                    continue
+
+            # Split the complete event (up to the end_of_field) into event_string,
+            # and retain anything after the current complete event in self.buf
+            # for next time.
+            (event_string, self.buf) = re.split(end_of_field, self.buf, maxsplit=1)
+            msg = Event.parse(event_string)
+
+            # If the server requests a specific retry delay, we need to honor it.
+            if msg.retry:
+                self.retry = msg.retry
+
+            # last_id should only be set if included in the message.  It's not
+            # forgotten if a message omits it.
+            if msg.id:
+                self.last_id = msg.id
+
+            #Set the last event ID string of the event source to the value of the last event ID buffer.
+            msg.lastEventId =self.last_id
+
+            # if data in event, emit and return
+            if msg.data !='':
+                return msg
+       
 
     if six.PY2:
         next = __next__


Test patch:
diff --git a/test_sseclient.py b/test_sseclient.py
index 73cf7a3..8ecef81 100644
--- a/test_sseclient.py
+++ b/test_sseclient.py
@@ -48,12 +48,23 @@ def test_no_space():
     m = E.parse('data:hi')
     assert m.data == 'hi'
 
+def test_with_space():
+    m = E.parse('data: hi')
+    assert m.data == 'hi'
+
+def test_with_leading_space():
+    m = E.parse('data:  hi')
+    assert m.data == ' hi'
 
 def test_comment():
     raw = ":this is a comment\ndata: this is some data"
     m = E.parse(raw)
     assert m.data == 'this is some data'
 
+def test_comment_only():
+    raw = ":this is a comment"
+    m = E.parse(raw)
+    assert m.data == ''
 
 def test_retry_is_integer():
     m = E.parse('data: hi\nretry: 4000')
@@ -73,9 +84,10 @@ def test_eols():
 
 
 class FakeResponse(object):
-    def __init__(self, status_code, content, headers=None):
+    def __init__(self, status_code, content, headers=None, encoding="utf-8"):
         self.status_code = status_code
-        self.encoding = "utf-8"
+        self.encoding = encoding
+        self.apparent_encoding = "utf-8"
         if not isinstance(content, six.text_type):
             content = content.decode("utf-8")
         self.stream = content
@@ -95,9 +107,10 @@ def join_events(*events):
 
 
 # Tests of parsing a multi event stream
-def test_last_id_remembered(monkeypatch):
+@pytest.mark.parametrize("encoding", ["utf-8", None])
+def test_last_id_remembered(monkeypatch, encoding):
     content = 'data: message 1\nid: abcdef\n\ndata: message 2\n\n'
-    fake_get = mock.Mock(return_value=FakeResponse(200, content))
+    fake_get = mock.Mock(return_value=FakeResponse(200, content, encoding=encoding))
     monkeypatch.setattr(requests, 'get', fake_get)
 
     c = sseclient.SSEClient('http://blah.com')
@@ -121,6 +134,26 @@ def test_retry_remembered(monkeypatch):
     assert m2.retry is None
     assert c.retry == 5000
 
+def test_commentonly_ignored(monkeypatch):
+    content = ':comment\n\ndata: message after comment\n\n'
+    fake_get = mock.Mock(return_value=FakeResponse(200, content))
+    monkeypatch.setattr(requests, 'get', fake_get)
+
+    c = sseclient.SSEClient('http://blah.com')
+    #the comment only event should be ignored entirely and not emitted
+    m1 = next(c)
+    assert m1.data == 'message after comment'
+
+def test_retryonly_ignored(monkeypatch):
+    content = 'retry: 10000\n\ndata:  will be emitted\n\n'
+    fake_get = mock.Mock(return_value=FakeResponse(200, content))
+    monkeypatch.setattr(requests, 'get', fake_get)
+
+    c = sseclient.SSEClient('http://blah.com')
+    #the retry only event should be processed but not emitted
+
+    m1 = next(c)
+    assert m1.data == ' will be emitted'
 
 def test_extra_newlines_after_event(monkeypatch):
     """
@@ -149,8 +182,8 @@ data: hello3
 
     assert m1.event == 'hello'
     assert m1.data == 'hello1'
-    assert m2.data == 'hello2'
     assert m2.event == 'hello'
+    assert m2.data == 'hello2'
     assert m3.data == 'hello3'
     assert m3.event == 'hello'