Long-running socket watcher, needs proper cleanup

mopidy_iris/core.py

@@ -63,7 +63,8 @@ class IrisCore(pykka.ThreadingActor):
         logger.info('Stopping Iris')
 
         if self.snapcast_daemon:
-            self.snapcast_daemon.stop()
+            logger.info('Stoppping Snapcast daemon')
+            self.snapcast_daemon.close()
 
 
     ##

mopidy_iris/snapcast.py

@@ -1,6 +1,6 @@
 
 from threading import Thread
-import os, logging, subprocess, socket, json, random, string
+import os, logging, subprocess, socket, json, random, string, select
 
 # import logger
 logger = logging.getLogger(__name__)
@@ -24,9 +24,9 @@ class IrisSnapcast(object):
         port = int(self.config['iris']['snapcast_port'])
 
         try:
-            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            self.socket.settimeout(10)
-            self.socket.connect((host, port))
+            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.sock.settimeout(10)
+            self.sock.connect((host, port))
             logger.debug("Snapcast connection established on "+host+":"+str(port))
 
         except socket.gaierror, e:
@@ -43,32 +43,39 @@ class IrisSnapcast(object):
     # @param broadcast = Method
     ##
     def listen(self, broadcast):
-        logger.info("Established Snapcast listener")
-
-        broadcast(data={'method':'snapcast_connected'})
 
+        self.connect()
         self.listen = True
 
+        logger.info("Established Snapcast listener")
+        broadcast(data={'method':'snapcast_connected'})
+
+        messages = []
         data = ""
+        select.select([], [self.sock], [])
 
         while self.listen:
 
-            # TODO: This works great, but when no data over X seconds we hit the timeout
-            # Possibly stop timeout? seems to lock things up
-            data = data + self.socket.recv(1024)
-            print data
-            if data.endswith(u"\r\n"):
+            readlist, writelist, exceptionlist = select.select([self.sock], [], [], 5)
 
-                message = data
-                data = ""
+            # Check if we've got any lists
+            if [readlist, writelist, exceptionlist] != [[], [], []]:
 
-                logger.debug("Incoming Snapcast message: "+message)
+                # Rread and print the available data on any of the read list
+                for socket in readlist:
 
-                try:
-                    message = json.loads(message)
-                    broadcast(data=message)
-                except:
-                    logger.error("Malformed Snapcast message: "+message)
+                    message = socket.recv(1024)
+
+                    try:
+                        message = json.loads(message)
+
+                        # Prefix with snapcast
+                        message['method'] = "snapcast_"+message['method']
+
+                        # Broadcast to all clients
+                        broadcast(data=message)
+                    except:
+                        logger.error("Malformed Snapcast message: "+message)
 
 
     ##
@@ -88,7 +95,7 @@ class IrisSnapcast(object):
     def request(self, data):
 
         # No socket to make request on
-        if self.socket == None:
+        if self.sock == None:
             raise Exception("Socket not established");
 
         # Construct our request, based on the provided data
@@ -107,7 +114,7 @@ class IrisSnapcast(object):
 
         # Attempt to send the request
         try:
-            self.socket.send(request.encode('ascii')+b"\n")
+            self.sock.send(request.encode('ascii')+b"\n")
 
         except socket.error, e:
             logger.error("Iris could not send request to Snapcast: %s" % e)
@@ -123,7 +130,7 @@ class IrisSnapcast(object):
         while True:
 
             try:
-                response = self.socket.recv(8192)
+                response = self.sock.recv(8192)
 
             except socket.error, e:
                 logger.error("Iris failed to receive Snapcast response: %s" % e)
@@ -143,13 +150,9 @@ class IrisSnapcast(object):
                 logger.debug(response)
 
                 if 'result' in response:
-                    socket_response = {
-                        'response': response['result']
-                    }
+                    socket_response = response['result']
                 else:
-                    socket_response = {
-                        'error': response['error']
-                    }
+                    socket_response = response['error']
 
             except:
                 logger.error("Iris received malformed Snapcast response: "+response)

mopidy_iris/snapcast_thread.py

@@ -8,6 +8,8 @@ logger = logging.getLogger(__name__)
 
 class IrisSnapcastThread(Thread):
 
+    sock = None
+
     def __init__(self, config, broadcast):
         Thread.__init__(self)
         self.config = config
@@ -19,7 +21,12 @@ class IrisSnapcastThread(Thread):
     ##
     def run(self):
 
-        socket = IrisSnapcast(self.config)
-        socket.connect()
-        socket.listen(self.broadcast)
-        
+        self.sock = IrisSnapcast(self.config)
+        self.sock.listen(self.broadcast)
+
+
+    ##
+    # Instruction to stop this thread
+    ##
+    def close(self):
+        self.sock.stop_listening()

mopidy_iris/static/app.js

@@ -21083,6 +21083,7 @@ exports.deleteClient = deleteClient;
 exports.setGroupStream = setGroupStream;
 exports.setGroupMute = setGroupMute;
 exports.setGroupVolume = setGroupVolume;
+exports.eventReceived = eventReceived;
 exports.serverLoaded = serverLoaded;
 exports.clientLoaded = clientLoaded;
 exports.clientsLoaded = clientsLoaded;
@@ -21170,6 +21171,14 @@ function setGroupVolume(id, percent) {
 	};
 }
 
+function eventReceived(message) {
+	return {
+		type: 'SNAPCAST_EVENT_RECEIVED',
+		method: message.method,
+		params: message.params
+	};
+}
+
 /**
  * Record loaders
  *
@@ -54274,6 +54283,7 @@ var pusherActions = __webpack_require__(14);
 var lastfmActions = __webpack_require__(21);
 var geniusActions = __webpack_require__(41);
 var spotifyActions = __webpack_require__(8);
+var snapcastActions = __webpack_require__(46);
 
 var PusherMiddleware = function () {
     var _this = this;
@@ -54417,6 +54427,11 @@ var PusherMiddleware = function () {
                         store.dispatch(uiActions.createNotification({ type: 'bad', content: 'Test failed' }));
                         break;
                 }
+
+                // Pass snapcast events to the Snapcast service
+                if (message.method.startsWith('snapcast_')) {
+                    store.dispatch(snapcastActions.eventReceived(message));
+                }
             }
         }
     };
@@ -61102,6 +61117,39 @@ var SnapcastMiddleware = function () {
 
                         break;
 
+                    case 'SNAPCAST_EVENT_RECEIVED':
+
+                        console.log(action);
+
+                        // Drop the prefix
+                        action.method = action.method.replace('snapcast_', '');
+
+                        switch (action.method) {
+
+                            case 'Client.OnVolumeChanged':
+                                store.dispatch(snapcastActions.clientLoaded({
+                                    id: action.params.id,
+                                    mute: action.params.volume.muted,
+                                    volume: action.params.volume.percent
+                                }));
+                                break;
+
+                            case 'Client.OnLatencyChanged':
+                                store.dispatch(snapcastActions.clientLoaded({
+                                    id: action.params.id,
+                                    latency: action.params.latency
+                                }));
+                                break;
+
+                            case 'Client.OnNameChanged':
+                                store.dispatch(snapcastActions.clientLoaded({
+                                    id: action.params.id,
+                                    name: action.params.name
+                                }));
+                                break;
+                        }
+                        break;
+
                     // This action is irrelevant to us, pass it on to the next middleware
                     default:
                         return next(action);