home-assistant · balloob · Sep 7, 2016 · Sep 5, 2016
diff --git a/homeassistant/components/api.py b/homeassistant/components/api.py
@@ -98,31 +98,32 @@ def forward_events(event):
 
         def stream():
             """Stream events to response."""
-            self.hass.bus.listen(MATCH_ALL, forward_events)
+            unsub_stream = self.hass.bus.listen(MATCH_ALL, forward_events)
 
-            _LOGGER.debug('STREAM %s ATTACHED', id(stop_obj))
+            try:
+                _LOGGER.debug('STREAM %s ATTACHED', id(stop_obj))
 
-            # Fire off one message right away to have browsers fire open event
-            to_write.put(STREAM_PING_PAYLOAD)
+                # Fire off one message so browsers fire open event right away
+                to_write.put(STREAM_PING_PAYLOAD)
 
-            while True:
-                try:
-                    payload = to_write.get(timeout=STREAM_PING_INTERVAL)
+                while True:
+                    try:
+                        payload = to_write.get(timeout=STREAM_PING_INTERVAL)
 
-                    if payload is stop_obj:
-                        break
+                        if payload is stop_obj:
+                            break
 
-                    msg = "data: {}\n\n".format(payload)
-                    _LOGGER.debug('STREAM %s WRITING %s', id(stop_obj),
-                                  msg.strip())
-                    yield msg.encode("UTF-8")
-                except queue.Empty:
-                    to_write.put(STREAM_PING_PAYLOAD)
-                except GeneratorExit:
-                    break
-
-            _LOGGER.debug('STREAM %s RESPONSE CLOSED', id(stop_obj))
-            self.hass.bus.remove_listener(MATCH_ALL, forward_events)
+                        msg = "data: {}\n\n".format(payload)
+                        _LOGGER.debug('STREAM %s WRITING %s', id(stop_obj),
+                                      msg.strip())
+                        yield msg.encode("UTF-8")
+                    except queue.Empty:
+                        to_write.put(STREAM_PING_PAYLOAD)
+                    except GeneratorExit:
+                        break
+            finally:
+                _LOGGER.debug('STREAM %s RESPONSE CLOSED', id(stop_obj))
+                unsub_stream()
 
         return self.Response(stream(), mimetype='text/event-stream')
 

@@ -13,7 +13,7 @@
 from homeassistant.components.light import is_on, turn_on
 from homeassistant.components.sun import next_setting, next_rising
 from homeassistant.components.switch import DOMAIN, SwitchDevice
-from homeassistant.const import CONF_NAME, CONF_PLATFORM, EVENT_TIME_CHANGED
+from homeassistant.const import CONF_NAME, CONF_PLATFORM
 from homeassistant.helpers.event import track_utc_time_change
 from homeassistant.util.color import color_temperature_to_rgb as temp_to_rgb
 from homeassistant.util.color import color_RGB_to_xy
@@ -124,7 +124,7 @@ def __init__(self, name, hass, state, lights, start_time, stop_time,
         self._stop_colortemp = stop_colortemp
         self._brightness = brightness
         self._mode = mode
-        self.tracker = None
+        self.unsub_tracker = None
 
     @property
     def name(self):
@@ -139,15 +139,17 @@ def is_on(self):
     def turn_on(self, **kwargs):
         """Turn on flux."""
         self._state = True
-        self.tracker = track_utc_time_change(self.hass,
-                                             self.flux_update,
-                                             second=[0, 30])
+        self.unsub_tracker = track_utc_time_change(self.hass, self.flux_update,
+                                                   second=[0, 30])
         self.update_ha_state()
 
     def turn_off(self, **kwargs):
         """Turn off flux."""
+        if self.unsub_tracker is not None:
+            self.unsub_tracker()
+            self.unsub_tracker = None
+
         self._state = False
-        self.hass.bus.remove_listener(EVENT_TIME_CHANGED, self.tracker)
         self.update_ha_state()
 
     # pylint: disable=too-many-locals

@@ -299,7 +299,7 @@ def listen(self, event_type, listener):
 
         def remove_listener():
             """Remove the listener."""
-            self.remove_listener(event_type, listener)
+            self._remove_listener(event_type, listener)
 
         return remove_listener
 
@@ -309,7 +309,7 @@ def listen_once(self, event_type, listener):
         To listen to all events specify the constant ``MATCH_ALL``
         as event_type.
 
-        Returns registered listener that can be used with remove_listener.
+        Returns function to unsubscribe the listener.
         """
         @ft.wraps(listener)
         def onetime_listener(event):
@@ -323,15 +323,21 @@ def onetime_listener(event):
             # This will make sure the second time it does nothing.
             setattr(onetime_listener, 'run', True)
 
-            self.remove_listener(event_type, onetime_listener)
+            remove_listener()
 
             listener(event)
 
-        self.listen(event_type, onetime_listener)
+        remove_listener = self.listen(event_type, onetime_listener)
 
-        return onetime_listener
+        return remove_listener
 
     def remove_listener(self, event_type, listener):
+        """Remove a listener of a specific event_type. (DEPRECATED 0.28)."""
+        _LOGGER.warning('bus.remove_listener has been deprecated. Please use '
+                        'the function returned from calling listen.')
+        self._remove_listener(event_type, listener)
+
+    def _remove_listener(self, event_type, listener):
         """Remove a listener of a specific event_type."""
         with self._lock:
             try:
@@ -344,7 +350,8 @@ def remove_listener(self, event_type, listener):
             except (KeyError, ValueError):
                 # KeyError is key event_type listener did not exist
                 # ValueError if listener did not exist within event_type
-                pass
+                _LOGGER.warning('Unable to remove unknown listener %s',
+                                listener)
 
 
 class State(object):
@@ -688,14 +695,13 @@ def service_executed(call):
                 if call.data[ATTR_SERVICE_CALL_ID] == call_id:
                     executed_event.set()
 
-            self._bus.listen(EVENT_SERVICE_EXECUTED, service_executed)
+            unsub = self._bus.listen(EVENT_SERVICE_EXECUTED, service_executed)
 
         self._bus.fire(EVENT_CALL_SERVICE, event_data)
 
         if blocking:
             success = executed_event.wait(SERVICE_CALL_LIMIT)
-            self._bus.remove_listener(
-                EVENT_SERVICE_EXECUTED, service_executed)
+            unsub()
             return success
 
     def _event_to_service_call(self, event):

@@ -211,16 +211,17 @@ def __init__(self, hass, restrict_origin=None):
         self._targets = {}
 
         self._lock = threading.Lock()
+        self._unsub_listener = None
 
     def connect(self, api):
         """Attach to a Home Assistant instance and forward events.
 
         Will overwrite old target if one exists with same host/port.
         """
         with self._lock:
-            if len(self._targets) == 0:
-                # First target we get, setup listener for events
-                self.hass.bus.listen(ha.MATCH_ALL, self._event_listener)
+            if self._unsub_listener is None:
+                self._unsub_listener = self.hass.bus.listen(
+                    ha.MATCH_ALL, self._event_listener)
 
             key = (api.host, api.port)
 
@@ -235,8 +236,8 @@ def disconnect(self, api):
 
             if len(self._targets) == 0:
                 # Remove event listener if no forwarding targets present
-                self.hass.bus.remove_listener(ha.MATCH_ALL,
-                                              self._event_listener)
+                self._unsub_listener()
+                self._unsub_listener = None
 
             return did_remove
 

diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
@@ -280,7 +280,7 @@ def test_platform_specific_config_validation(self):
 
         loader.set_component(
             'switch.platform_a',
-            MockPlatform('comp_b', platform_schema=platform_schema))
+            MockPlatform(platform_schema=platform_schema))
 
         assert not bootstrap.setup_component(self.hass, 'switch', {
             'switch': {

diff --git a/tests/test_core.py b/tests/test_core.py
@@ -166,14 +166,14 @@ def listener(_): pass
         self.assertEqual(old_count + 1, len(self.bus.listeners))
 
         # Try deleting a non registered listener, nothing should happen
-        self.bus.remove_listener('test', lambda x: len)
+        self.bus._remove_listener('test', lambda x: len)
 
         # Remove listener
-        self.bus.remove_listener('test', listener)
+        self.bus._remove_listener('test', listener)
         self.assertEqual(old_count, len(self.bus.listeners))
 
         # Try deleting listener while category doesn't exist either
-        self.bus.remove_listener('test', listener)
+        self.bus._remove_listener('test', listener)
 
     def test_unsubscribe_listener(self):
         """Test unsubscribe listener from returned function."""