Fix bug in nilmdb.stream_remove with max_removals

Instead of returning a single number at the end of N nilmdb calls, we
now use a generator that returns one line of text every time there's a
new count of rows removed.  This ensures that the connection will stay
alive for very long removals.

nilmdb/client/client.py

@@ -122,7 +122,10 @@ class Client(object):
             params["start"] = timestamp_to_string(start)
         if end is not None:
             params["end"] = timestamp_to_string(end)
-        return self.http.post("stream/remove", params)
+        total = 0
+        for count in self.http.post_gen("stream/remove", params):
+            total += int(count)
+        return total
 
     @contextlib.contextmanager
     def stream_insert_context(self, path, start = None, end = None):

nilmdb/client/httpclient.py

@@ -137,5 +137,14 @@ class HTTPClient(object):
         """Simple GET (parameters in URL) returning a generator"""
         return self._req_gen("GET", url, params, binary = binary)
 
+    def post_gen(self, url, params = None):
+        """Simple POST (parameters in body) returning a generator"""
+        if self.post_json:
+            return self._req_gen("POST", url, None,
+                                 json.dumps(params),
+                                 { 'Content-type': 'application/json' })
+        else:
+            return self._req_gen("POST", url, None, params)
+
     # Not much use for a POST or PUT generator, since they don't
     # return much data.

nilmdb/server/nilmdb.py

@@ -675,6 +675,7 @@ class NilmDB(object):
 
             # Count how many were removed
             removed += row_end - row_start
+            remaining -= row_end - row_start
 
             if restart is not None:
                 break

nilmdb/server/server.py

@@ -347,24 +347,34 @@ class Stream(NilmApp):
     # /stream/remove?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
     @cherrypy.tools.json_in()
-    @cherrypy.tools.json_out()
-    @exception_to_httperror(NilmDBError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
+    @chunked_response
+    @response_type("application/x-json-stream")
     def remove(self, path, start = None, end = None):
         """
         Remove data from the backend database.  Removes all data in
-        the interval [start, end).  Returns the number of data points
-        removed.
+        the interval [start, end).
+
+        Returns the number of data points removed.  Since this is a potentially
+        long-running operation, multiple numbers may be returned as the
+        data gets removed from the backend database.  The total number of
+        points removed is the sum of all of these numbers.
         """
         (start, end) = self._get_times(start, end)
-        total_removed = 0
-        while True:
-            (removed, restart) = self.db.stream_remove(path, start, end)
-            total_removed += removed
-            if restart is None:
-                break
-            start = restart
-        return total_removed
+
+        if len(self.db.stream_list(path = path)) != 1:
+            raise cherrypy.HTTPError("404", "No such stream: " + path)
+
+        @workaround_cp_bug_1200
+        def content(start, end):
+            # Note: disable chunked responses to see tracebacks from here.
+            while True:
+                (removed, restart) = self.db.stream_remove(path, start, end)
+                yield json.dumps(removed) + "\r\n"
+                if restart is None:
+                    break
+                start = restart
+        return content(start, end)
 
     # /stream/intervals?path=/newton/prep
     # /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0