Bus and StateMachine are now thread-safe

homeassistant/__init__.py

@@ -158,10 +158,13 @@ class Bus(object):
         self._event_listeners = {}
         self._services = {}
         self.logger = logging.getLogger(__name__)
+        self.event_lock = threading.Lock()
+        self.service_lock = threading.Lock()
 
     @property
     def services(self):
         """ Dict with per domain a list of available services. """
+        with self.service_lock:
             return {domain: self._services[domain].keys()
                     for domain in self._services}
 
@@ -170,12 +173,13 @@ class Bus(object):
         """ Dict with events that is being listened for and the number
         of listeners.
         """
+        with self.event_lock:
             return {key: len(self._event_listeners[key])
                     for key in self._event_listeners}
 
     def call_service(self, domain, service, service_data=None):
         """ Calls a service. """
-
+        with self.service_lock:
             try:
                 self._services[domain][service]
             except KeyError:
@@ -183,8 +187,7 @@ class Bus(object):
                 raise ServiceDoesNotExistException(
                     "Service does not exist: {}/{}".format(domain, service))
 
-        if not service_data:
-            service_data = {}
+            service_data = service_data or {}
 
             def run():
                 """ Executes a service. """
@@ -193,14 +196,16 @@ class Bus(object):
                 try:
                     self._services[domain][service](service_call)
                 except Exception:  # pylint: disable=broad-except
-                self.logger.exception("Bus:Exception in service {}/{}".format(
+                    self.logger.exception(
+                        "Bus:Exception in service {}/{}".format(
                             domain, service))
 
-        # We dont want the eventbus to be blocking - run in a thread.
+            # We dont want the bus to be blocking - run in a thread.
             threading.Thread(target=run).start()
 
     def register_service(self, domain, service, service_callback):
         """ Register a service. """
+        with self.service_lock:
             try:
                 self._services[domain][service] = service_callback
             except KeyError:
@@ -209,7 +214,7 @@ class Bus(object):
 
     def fire_event(self, event_type, event_data=None):
         """ Fire an event. """
-
+        with self.event_lock:
             # Copy the list of the current listeners because some listeners
             # choose to remove themselves as a listener while being executed
             # which causes the iterator to be confused.
@@ -231,9 +236,9 @@ class Bus(object):
                 for listener in listeners:
                     try:
                         listener(event)
-
                     except Exception:  # pylint: disable=broad-except
-                    self.logger.exception("Bus:Exception in event listener")
+                        self.logger.exception(
+                            "Bus:Exception in event listener")
 
             # We dont want the bus to be blocking - run in a thread.
             threading.Thread(target=run).start()
@@ -244,6 +249,7 @@ class Bus(object):
         To listen to all events specify the constant ``MATCH_ALL``
         as event_type.
         """
+        with self.event_lock:
             try:
                 self._event_listeners[event_type].append(listener)
             except KeyError:  # event_type did not exist
@@ -257,7 +263,6 @@ class Bus(object):
 
         Note: at the moment it is impossible to remove a one time listener.
         """
-
         def onetime_listener(event):
             """ Removes listener from eventbus and then fires listener. """
             self.remove_event_listener(event_type, onetime_listener)
@@ -268,13 +273,17 @@ class Bus(object):
 
     def remove_event_listener(self, event_type, listener):
         """ Removes a listener of a specific event_type. """
+        with self.event_lock:
             try:
                 self._event_listeners[event_type].remove(listener)
 
-            if len(self._event_listeners[event_type]) == 0:
+                # delete event_type list if empty
+                if not self._event_listeners[event_type]:
                     del self._event_listeners[event_type]
 
             except (KeyError, ValueError):
+                # KeyError is key event_type did not exist
+                # ValueError if the list [event_type] did not contain listener
                 pass
 
 
@@ -346,12 +355,14 @@ class StateMachine(object):
     @property
     def categories(self):
         """ List of categories which states are being tracked. """
+        with self.lock:
             return self.states.keys()
 
     def remove_category(self, category):
         """ Removes a category from the state machine.
 
         Returns boolean to indicate if a category was removed. """
+        with self.lock:
             try:
                 del self.states[category]
 
@@ -368,8 +379,7 @@ class StateMachine(object):
 
         attributes = attributes or {}
 
-        self.lock.acquire()
-
+        with self.lock:
             # Add category if it does not exist
             if category not in self.states:
                 self.states[category] = State(new_state, attributes)
@@ -388,11 +398,10 @@ class StateMachine(object):
                                          'old_state': old_state,
                                          'new_state': self.states[category]})
 
-        self.lock.release()
-
     def get_state(self, category):
         """ Returns a dict (state, last_changed, attributes) describing
             the state of the specified category. """
+        with self.lock:
             try:
                 # Make a copy so people won't mutate the state
                 return self.states[category].copy()
@@ -403,9 +412,11 @@ class StateMachine(object):
 
     def is_state(self, category, state):
         """ Returns True if category exists and is specified state. """
-        cur_state = self.get_state(category)
-
-        return cur_state and cur_state.state == state
+        try:
+            return self.get_state(category).state == state
+        except AttributeError:
+            # get_state returned None
+            return False
 
 
 class Timer(threading.Thread):