Listen to incoming messages in a new thread

README.md

@@ -86,6 +86,12 @@ logging.info(app)
 info = tv.rest_device_info()
 logging.info(info)
 
+# Listen to incoming events
+def handle_message(msg):
+    print(msg['event'])
+
+tv.events.subscribe(const.Events.ALL, handle_message)
+
 ```
 
 ## License

samsungtvws/const.py

@@ -0,0 +1,7 @@
+
+class Events:
+    ALL = '*'
+    APP_LIST = 'ed.installedApp.get'
+    APP_LAUNCH = 'ed.apps.launch'
+    CLIENT_CONNECT = 'ms.channel.clientConnect'
+    CLIENT_DISCONNECT = 'ms.channel.clientDisconnect'

samsungtvws/remote.py

@@ -22,12 +22,16 @@
 import base64
 import json
 import logging
+import threading
 import time
 import ssl
 import websocket
 import requests
 from . import exceptions
 from . import shortcuts
+import requests
+from json import JSONDecodeError
+from .simple_pub_sub import SimplePubSub
 
 _LOGGING = logging.getLogger(__name__)
 
@@ -47,6 +51,12 @@ def __init__(self, host, token=None, token_file=None, port=8001, timeout=None, k
         self.key_press_delay = key_press_delay
         self.name = name
         self.connection = None
+        self._app_list = None
+        self._app_list_fetch_in_progress = False
+        self.listen_thread = None
+        self.should_continue_listening = True
+        self.events = SimplePubSub()
+        self.events.subscribe('ed.installedApp.get', self._handle_app_list_event)
 
     def __enter__(self):
         return self
@@ -103,6 +113,10 @@ def _set_token(self, token):
         else:
             self.token = token
 
+    def _handle_app_list_event(self, new_app_list):
+        self._app_list = new_app_list.get('data').get('data')
+        self._app_list_fetch_in_progress = False
+
     def _ws_send(self, command, key_press_delay=None):
         if self.connection is None:
             self.open()
@@ -135,6 +149,10 @@ def _process_api_response(self, response):
             raise exceptions.ResponseError('Failed to parse response from TV. Maybe feature not supported on this model')
 
     def open(self):
+        if self.connection:
+            # someone else already created a new connection
+            return
+
         is_ssl = self._is_ssl_connection()
         url = self._format_websocket_url(is_ssl)
         sslopt = {'cert_reqs': ssl.CERT_NONE} if is_ssl else {}
@@ -156,12 +174,41 @@ def open(self):
             self.close()
             raise exceptions.ConnectionFailure(response)
 
+        self._do_after_connect()
+
+    def _start_listening(self):
+        self.listen_thread = threading.Thread(target=self._listen_to_messages_loop)
+        self.listen_thread.start()
+
+    def _listen_to_messages_loop(self):
+        if not self.connection:
+            self.open()
+
+        while self.connection and self.should_continue_listening:
+            raw_msg = None
+            try:
+                raw_msg = self.connection.recv()
+                msg = json.loads(raw_msg)
+                if 'event' in msg:
+                    topic = msg['event']
+                    data = msg
+                    self.events.publish(topic, data)
+                else:
+                    self.events.publish('*', msg)
+
+            except ValueError as ex:
+                print("Error parsing message %s. Error: %s", raw_msg, ex)
+            except Exception as ex:
+                print("Error reading message from TV. Error: %s", ex)
+                time.sleep(3)
+
     def close(self):
         if self.connection:
             self.connection.close()
 
         self.connection = None
         _LOGGING.debug('Connection closed.')
+        self.listen_thread.join()
 
     def send_key(self, key, times=1, key_press_delay=None, cmd='Click'):
         for _ in range(times):
@@ -226,8 +273,8 @@ def open_browser(self, url):
             url
         )
 
-    def app_list(self):
-        _LOGGING.debug('Get app list')
+    def request_app_list(self):
+        _LOGGING.info('request app list')
         self._ws_send({
             'method': 'ms.channel.emit',
             'params': {
@@ -236,11 +283,16 @@ def app_list(self):
             }
         })
 
-        response = self._process_api_response(self.connection.recv())
-        if response.get('data') and response.get('data').get('data'):
-            return response.get('data').get('data')
-        else:
-            return response
+    def app_list(self):
+        self._app_list_fetch_in_progress = True
+        self.request_app_list()
+
+        attemps_left = 10
+        while self._app_list_fetch_in_progress and attemps_left:
+            time.sleep(1)
+            attemps_left -= 1
+
+        return self._app_list
 
     def rest_device_info(self):
         _LOGGING.debug('Get device info via rest api')
@@ -269,3 +321,6 @@ def rest_app_install(self, app_id):
 
     def shortcuts(self):
         return shortcuts.SamsungTVShortcuts(self)
+
+    def _do_after_connect(self):
+        self._start_listening()

samsungtvws/simple_pub_sub.py

@@ -0,0 +1,35 @@
+
+class ListenerEntry(object):
+    def __init__(self, callable):
+        self.callable = callable
+
+class SimplePubSub(object):
+    def __init__(self):
+        self.topics = {
+            '*': set()
+        }
+
+    def subscribe(self, topic, callable):
+        if topic not in self.topics:
+            self.topics[topic] = set()
+
+        self.topics[topic].add(ListenerEntry(callable))
+
+    def unsubscribe(self, topic, callable):
+        if topic in self.topics:
+            listeners = self.topics[topic]
+            listeners.remove(callable)
+            if not len(listeners):
+                del self.topics[topic]
+        else:
+            raise ValueError("unsubscribe: topic does not exists")
+
+    def publish(self, topic, data):
+        self._publish(topic, data)
+        if topic != '*':
+            self._publish('*', data)
+
+    def _publish(self, topic, data):
+        if topic in self.topics:
+            for l in self.topics[topic]:
+                l.callable(data)