Merge 72c08e6 into a41f81e

cloudant/database.py

@@ -1,3 +1,4 @@
+import json
 from .resource import Resource
 from .document import Document
 from .design import Design
@@ -82,12 +83,25 @@ def changes(self, **kwargs):
         For more information about the `_changes` feed, see
         [the docs](http://docs.cloudant.com/api/database.html#obtaining-a-list-of-changes).
         """
+        size = 512  # requests default chunk size value
         if 'params' in kwargs:
             if 'feed' in kwargs['params']:
                 if kwargs['params']['feed'] == 'continuous':
                     kwargs['stream'] = True
+                    size = 1  # 1 byte because we don't want to hold the last
+                              # record in memory buffer in awaiting for new data
 
-        return self.get('_changes', **kwargs)
+        response = self.get('_changes', **kwargs)
+        response.raise_for_status()
+
+        # TODO: this code duplicates index.Index.__iter__
+        for line in response.iter_lines(chunk_size=size):
+            if not line:
+                continue
+            line = line.decode('utf-8')
+            if line[-1] == ',':
+                line = line[:-1]
+            yield json.loads(line)
 
     def missing_revs(self, revs, **kwargs):
         """

test/__init__.py

@@ -1,5 +1,7 @@
 import cloudant
 from collections import defaultdict
+from types import GeneratorType
+import time
 import unittest
 import os
 
@@ -166,10 +168,28 @@ def testAllDocsWithKeys(self):
         assert resp.status_code == 200
 
     def testChanges(self):
-        assert self.db.changes().status_code == 200
-        assert self.db.changes(params={
-            'feed': 'continuous'
-        }).status_code == 200
+        assert isinstance(self.db.changes(), GeneratorType)
+
+    def testChangesContinuous(self):
+        def iterator(iterable):
+            _iterable = iter(iterable)
+            def wrapper():
+                for item in _iterable:
+                    return item
+            return wrapper
+
+        self.db.bulk_docs(self.test_doc, self.test_otherdoc)
+
+        resp = self.db.changes(params={
+            'feed': 'continuous',
+            'timeout': 2000
+        })
+        start = time.time()
+        _next = iterator(resp)
+        _next()
+        _next()
+        assert time.time() - start < 2, \
+            'should return changes event for both documents before block'
 
     def testViewCleanup(self):
         assert self.db.view_cleanup().status_code == 202