Get EventStream working

homeassistant/components/api.py

@@ -73,102 +73,94 @@ class APIEventStream(HomeAssistantView):
 
     def get(self, request):
         """Provide a streaming interface for the event bus."""
+        from eventlet.queue import Empty
         import eventlet
-        from eventlet import queue as eventlet_queue
+        import homeassistant.util.eventlet as eventlet_util
-        import queue as thread_queue
-        from threading import Event
-        from time import time
 
-        to_write = thread_queue.Queue()
+        cur_hub = eventlet.hubs.get_hub()
-        # to_write = eventlet.Queue()
+        request.environ['eventlet.minimum_write_chunk_size'] = 0
+
+        to_write = eventlet.Queue()
         stop_obj = object()
-        hass = self.hass
+        attached_ping = None
-        connection_closed = Event()
 
         restrict = request.args.get('restrict')
         if restrict:
             restrict = restrict.split(',')
 
-        restrict = False
+        def thread_ping(now):
+            """Called from time thread to add ping to queue."""
+            _LOGGER.debug('STREAM %s PING', id(stop_obj))
+            eventlet_util.spawn(cur_hub, to_write.put, STREAM_PING_PAYLOAD)
 
-        def ping(now):
+        def thread_forward_events(event):
-            """Add a ping message to queue."""
-            print(id(stop_obj), 'ping')
-            to_write.put(STREAM_PING_PAYLOAD)
-
-        def forward_events(event):
             """Forward events to the open request."""
-            print(id(stop_obj), 'forwarding', event)
             if event.event_type == EVENT_TIME_CHANGED:
-                pass
+                return
-            elif event.event_type == EVENT_HOMEASSISTANT_STOP:
+
-                to_write.put(stop_obj)
+            _LOGGER.debug('STREAM %s FORWARDING %s', id(stop_obj), event)
+
+            if event.event_type == EVENT_HOMEASSISTANT_STOP:
+                data = stop_obj
             else:
-                to_write.put(json.dumps(event, cls=rem.JSONEncoder))
+                data = json.dumps(event, cls=rem.JSONEncoder)
+
+            eventlet_util.spawn(cur_hub, to_write.put, data)
+
+        def cleanup():
+            """Clean up HA listeners."""
+            _LOGGER.debug("STREAM %s CLEANING UP", id(stop_obj))
+            self.hass.bus.remove_listener(EVENT_TIME_CHANGED, attached_ping)
+
+            if restrict:
+                for event in restrict:
+                    self.hass.bus.remove_listener(event, thread_forward_events)
+            else:
+                self.hass.bus.remove_listener(MATCH_ALL, thread_forward_events)
 
         def stream():
             """Stream events to response."""
+            nonlocal attached_ping
+
             if restrict:
                 for event_type in restrict:
-                    hass.bus.listen(event_type, forward_events)
+                    self.hass.bus.listen(event_type, thread_forward_events)
             else:
-                hass.bus.listen(MATCH_ALL, forward_events)
+                self.hass.bus.listen(MATCH_ALL, thread_forward_events)
 
             attached_ping = track_utc_time_change(
-                hass, ping, second=(0, 30))
+                self.hass, thread_ping, second=range(0, 60, 3)) #(0, 30))
 
-            print(id(stop_obj), 'attached goodness')
+            _LOGGER.debug('STREAM %s ATTACHED', id(stop_obj))
 
-            while not connection_closed.is_set():
+            while True:
                 try:
-                    print(id(stop_obj), "Try getting obj")
+                    # Somehow our queue.get takes too long to
-                    payload = to_write.get(False)
+                    # be notified of arrival of object. Probably
+                    # because of our spawning on hub in other thread
+                    # hack. Because current goal is to get this out,
+                    # We just timeout every second because it will
+                    # return right away if qsize() > 0.
+                    # So yes, we're basically polling :(
+                    # socket.io anyone?
+                    payload = to_write.get(timeout=1)
 
                     if payload is stop_obj:
                         break
 
                     msg = "data: {}\n\n".format(payload)
-                    print(id(stop_obj), msg)
+                    _LOGGER.debug('STREAM %s WRITING %s', id(stop_obj),
+                                  msg.strip())
                     yield msg.encode("UTF-8")
-                except eventlet_queue.Empty:
+                except Empty:
-                    print(id(stop_obj), "queue empty, sleep 0.5")
-                    eventlet.sleep(.5)
-                except GeneratorExit:
                     pass
+                except GeneratorExit:
+                    _LOGGER.debug('STREAM %s RESPONSE CLOSED', id(stop_obj))
+                    break
 
-            print(id(stop_obj), "cleaning up")
+            cleanup()
 
-            hass.bus.remove_listener(EVENT_TIME_CHANGED, attached_ping)
+        return self.Response(stream(), mimetype='text/event-stream')
-
-            if restrict:
-                for event in restrict:
-                    hass.bus.remove_listener(event, forward_events)
-            else:
-                hass.bus.remove_listener(MATCH_ALL, forward_events)
-
-        resp = self.Response(stream(), mimetype='text/event-stream')
-
-        def closing():
-            print()
-            print()
-            print()
-            print()
-            print()
-            print()
-            print()
-            print()
-            print(id(stop_obj), "CLOSING RESPONSE")
-            print()
-            print()
-            print()
-            print()
-            print()
-            print()
-            print()
-            connection_closed.set()
-
-        resp.call_on_close(closing)
-        return resp
 
 
 class APIConfigView(HomeAssistantView):

homeassistant/components/http.py

@@ -213,9 +213,6 @@ class HomeAssistantWSGI(object):
         """Register a folder to serve as a static path."""
         from static import Cling
 
-        if url_root in self.extra_apps:
-            _LOGGER.warning("Static path '%s' is being overwritten", path)
-
         headers = []
 
         if not self.development:
@@ -228,7 +225,14 @@ class HomeAssistantWSGI(object):
                 "public, max-age={}".format(cache_time)
             })
 
-        self.extra_apps[url_root] = Cling(path, headers=headers)
+        self.register_wsgi_app(url_root, Cling(path, headers=headers))
+
+    def register_wsgi_app(self, url_root, app):
+        """Register a path to serve a WSGI app."""
+        if url_root in self.extra_apps:
+            _LOGGER.warning("Url root '%s' is being overwritten", url_root)
+
+        self.extra_apps[url_root] = app
 
     def start(self):
         """Start the wsgi server."""
@@ -248,6 +252,7 @@ class HomeAssistantWSGI(object):
         )
         from werkzeug.routing import RequestRedirect
 
+        with request:
             adapter = self.url_map.bind_to_environ(request.environ)
             try:
                 endpoint, values = adapter.match()

homeassistant/util/eventlet.py

@@ -0,0 +1,9 @@
+"""Eventlet util methods."""
+
+
+def spawn(hub, func, *args, **kwargs):
+    """Spawns a function on specified hub."""
+    import eventlet
+    g = eventlet.greenthread.GreenThread(hub.greenlet)
+    hub.schedule_call_global(0, g.switch, func, args, kwargs)
+    return g

tests/components/test_api.py

@@ -435,52 +435,47 @@ class TestAPI(unittest.TestCase):
             headers=HA_HEADERS)
         self.assertEqual(200, req.status_code)
 
-    # def test_stream(self):
+    def test_stream(self):
-    #     """Test the stream."""
+        """Test the stream."""
-    #     listen_count = self._listen_count()
+        listen_count = self._listen_count()
-    #     with closing(requests.get(_url(const.URL_API_STREAM),
+        with closing(requests.get(_url(const.URL_API_STREAM),
-    #                               stream=True, headers=HA_HEADERS)) as req:
+                                  stream=True, headers=HA_HEADERS)) as req:
 
-    #         self.assertEqual(listen_count + 1, self._listen_count())
+            self.assertEqual(listen_count + 2, self._listen_count())
 
-    #         # eventlet.sleep(1)
+            hass.bus.fire('test_event')
-    #         print('firing event')
+            hass.pool.block_till_done()
 
-    #         hass.bus.fire('test_event')
+            data = self._stream_next_event(req)
-    #         hass.pool.block_till_done()
 
-    #         data = self._stream_next_event(req)
+            self.assertEqual('test_event', data['event_type'])
 
-    #         self.assertEqual('test_event', data['event_type'])
+    def test_stream_with_restricted(self):
+        """Test the stream with restrictions."""
+        listen_count = self._listen_count()
+        url = _url('{}?restrict=test_event1,test_event3'.format(
+            const.URL_API_STREAM))
 
-    # def test_stream_with_restricted(self):
+        with closing(requests.get(url, stream=True,
-    #     """Test the stream with restrictions."""
+                                  headers=HA_HEADERS)) as req:
-    #     listen_count = self._listen_count()
-    #     with closing(requests.get(_url(const.URL_API_STREAM),
-    #                               data=json.dumps({
-    #                                   'restrict':
-    #                                   'test_event1,test_event3'}),
-    #                               stream=True, headers=HA_HEADERS)) as req:
 
-    #         data = self._stream_next_event(req)
+            self.assertEqual(listen_count + 3, self._listen_count())
-    #         self.assertEqual('ping', data)
 
-    #         self.assertEqual(listen_count + 2, self._listen_count())
+            hass.bus.fire('test_event1')
+            hass.pool.block_till_done()
+            hass.bus.fire('test_event2')
+            hass.pool.block_till_done()
+            hass.bus.fire('test_event3')
+            hass.pool.block_till_done()
 
-    #         hass.bus.fire('test_event1')
+            data = self._stream_next_event(req)
-    #         hass.pool.block_till_done()
+            self.assertEqual('test_event1', data['event_type'])
-    #         hass.bus.fire('test_event2')
+            data = self._stream_next_event(req)
-    #         hass.pool.block_till_done()
+            self.assertEqual('test_event3', data['event_type'])
-    #         hass.bus.fire('test_event3')
-    #         hass.pool.block_till_done()
-
-    #         data = self._stream_next_event(req)
-    #         self.assertEqual('test_event1', data['event_type'])
-    #         data = self._stream_next_event(req)
-    #         self.assertEqual('test_event3', data['event_type'])
 
     def _stream_next_event(self, stream):
-        """Test the stream for next event."""
+        """Read the stream for next event while ignoring ping."""
+        while True:
             data = b''
             last_new_line = False
             for dat in stream.iter_content(1):
@@ -491,7 +486,10 @@ class TestAPI(unittest.TestCase):
 
             conv = data.decode('utf-8').strip()[6:]
 
-        return conv if conv == 'ping' else json.loads(conv)
+            if conv != 'ping':
+                break
+
+        return json.loads(conv)
 
     def _listen_count(self):
         """Return number of event listeners."""