Nonblocking update listener

couchpotato/core/notifications/core/main.py

@@ -1,5 +1,5 @@
 from couchpotato import get_session
-from couchpotato.api import addApiView
+from couchpotato.api import addApiView, addNonBlockApiView
 from couchpotato.core.event import addEvent
 from couchpotato.core.helpers.encoding import toUnicode
 from couchpotato.core.helpers.request import jsonified, getParam
@@ -8,14 +8,19 @@ from couchpotato.core.logger import CPLog
 from couchpotato.core.notifications.base import Notification
 from couchpotato.core.settings.model import Notification as Notif
 from sqlalchemy.sql.expression import or_
+import threading
 import time
+import uuid
 
 log = CPLog(__name__)
 
 
 class CoreNotifier(Notification):
 
+    m_lock = threading.RLock()
     messages = []
+    listeners = []
+
     listen_to = [
         'movie.downloaded', 'movie.snatched',
         'updater.available', 'updater.updated',
@@ -46,8 +51,17 @@ class CoreNotifier(Notification):
 }"""}
         })
 
+        addNonBlockApiView('notification.listener', (self.addListener, self.removeListener))
         addApiView('notification.listener', self.listener)
 
+
+        def test():
+            while True:
+                time.sleep(1)
+
+        addEvent('app.load', test)
+
+
     def markAsRead(self):
         ids = [x.strip() for x in getParam('ids').split(',')]
 
@@ -107,25 +121,79 @@ class CoreNotifier(Notification):
         ndict = n.to_dict()
         ndict['type'] = 'notification'
         ndict['time'] = time.time()
-        self.messages.append(ndict)
+
+        self.frontend(type = listener, data = data)
 
         #db.close()
         return True
 
     def frontend(self, type = 'notification', data = {}):
-        self.messages.append({
+
+        self.m_lock.acquire()
+        message = {
+            'id': str(uuid.uuid4()),
             'time': time.time(),
             'type': type,
             'data': data,
-        })
+        }
+        self.messages.append(message)
+
+        while True and not self.shuttingDown():
+            try:
+                listener, last_id = self.listeners.pop()
+                listener({
+                    'success': True,
+                    'result': [message],
+                })
+            except:
+                break
+
+        self.m_lock.release()
+
+        self.cleanMessages()
+
+    def addListener(self, callback, last_id = None):
+
+        if last_id:
+            messages = self.getMessages(last_id)
+            if len(messages) > 0:
+                return callback({
+                    'success': True,
+                    'result': messages,
+                })
+
+        self.listeners.append((callback, last_id))
+
+    def removeListener(self, callback):
+        for list_tuple in self.listeners:
+            try:
+                listener, last_id = list_tuple
+                if listener == callback:
+                    self.listeners.remove(list_tuple)
+            except:
+                pass
+
+    def cleanMessages(self):
+
+        for message in self.messages:
+            if message['time'] < (time.time() - 15):
+                self.messages.remove(message)
+
+    def getMessages(self, last_id):
+        self.m_lock.acquire()
+        recent = []
+        index = 0
+        for i in xrange(len(self.messages)):
+            index = len(self.messages) - i - 1
+            if self.messages[index]["id"] == last_id: break
+            recent = self.messages[index + 1:]
+
+        self.m_lock.release()
+        return recent or []
 
     def listener(self):
 
         messages = []
-        for message in self.messages:
-            #delete message older then 15s
-            if message['time'] > (time.time() - 15):
-                messages.append(message)
 
         # Get unread
         if getParam('init'):
@@ -139,9 +207,6 @@ class CoreNotifier(Notification):
                 ndict['type'] = 'notification'
                 messages.append(ndict)
 
-            #db.close()
-
-        self.messages = []
         return jsonified({
             'success': True,
             'result': messages,

couchpotato/core/notifications/core/static/notification.js

@@ -8,8 +8,7 @@ var NotificationBase = new Class({
 		self.setOptions(options);
 
 		// Listener
-		App.addEvent('load', self.startInterval.bind(self));
-		App.addEvent('unload', self.stopTimer.bind(self));
+		App.addEvent('unload', self.stopPoll.bind(self));
 		App.addEvent('notification', self.notify.bind(self));
 
 		// Add test buttons to settings page
@@ -30,7 +29,11 @@ var NotificationBase = new Class({
 				'href': App.createUrl('notifications'),
 				'text': 'Show older notifications'
 			})); */
-		})
+		});
+
+		window.addEvent('load', function(){
+			self.startInterval()
+		});
 
 	},
 
@@ -85,35 +88,55 @@ var NotificationBase = new Class({
 
 	startInterval: function(){
 		var self = this;
+		
+		if(self.stopped) return;
 
-		self.request = Api.request('notification.listener', {
-			'initialDelay': 100,
-    		'delay': 1500,
+		Api.request('notification.listener', {
     		'data': {'init':true},
     		'onSuccess': self.processData.bind(self)
-		})
-
-		self.request.startTimer()
+		}).send()
 
 	},
 
-	startTimer: function(){
-		if(this.request)
-			this.request.startTimer()
+	startPoll: function(){
+		var self = this;
+
+		if(self.stopped || (self.request && self.request.isRunning()))
+			return;
+
+		self.request = Api.request('nonblock/notification.listener', {
+    		'onSuccess': self.processData.bind(self),
+    		'data': {
+    			'last_id': self.last_id
+    		},
+    		'onFailure': function(){
+    			self.startPoll.delay(2000, self)
+    		}
+		}).send()
+
 	},
 
-	stopTimer: function(){
+	stopPoll: function(){
 		if(this.request)
-			this.request.stopTimer()
+			this.request.cancel()
+		this.stopped = true;
 	},
 
 	processData: function(json){
 		var self = this;
 
-		self.request.options.data = {}
-		Array.each(json.result, function(result){
-			App.fireEvent(result.type, result)
-		})
+
+		// Process data
+		if(json){
+			Array.each(json.result, function(result){
+				App.fireEvent(result.type, result)
+			})
+
+			self.last_id = json.result.getLast().id
+		}
+
+		// Restart poll
+		self.startPoll()
 	},
 
 	addTestButtons: function(){