Better locking for longpoll
This commit is contained in:
+10
-5
@@ -15,19 +15,24 @@ class NonBlockHandler(RequestHandler):
|
|||||||
|
|
||||||
@asynchronous
|
@asynchronous
|
||||||
def get(self, route):
|
def get(self, route):
|
||||||
|
cls = NonBlockHandler
|
||||||
start, stop = api_nonblock[route]
|
start, stop = api_nonblock[route]
|
||||||
self.stoppers.append(stop)
|
cls.stoppers.append(stop)
|
||||||
|
|
||||||
start(self.on_new_messages, last_id = self.get_argument("last_id", None))
|
start(self.onNewMessage, last_id = self.get_argument("last_id", None))
|
||||||
|
|
||||||
def on_new_messages(self, response):
|
def onNewMessage(self, response):
|
||||||
if self.request.connection.stream.closed():
|
if self.request.connection.stream.closed():
|
||||||
return
|
return
|
||||||
self.finish(response)
|
self.finish(response)
|
||||||
|
|
||||||
def on_connection_close(self):
|
def on_connection_close(self):
|
||||||
for stop in self.stoppers:
|
cls = NonBlockHandler
|
||||||
stop(self.on_new_messages)
|
|
||||||
|
for stop in cls.stoppers:
|
||||||
|
stop(self.onNewMessage)
|
||||||
|
|
||||||
|
cls.stoppers = []
|
||||||
|
|
||||||
|
|
||||||
def addApiView(route, func, static = False, docs = None, **kwargs):
|
def addApiView(route, func, static = False, docs = None, **kwargs):
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ log = CPLog(__name__)
|
|||||||
|
|
||||||
class CoreNotifier(Notification):
|
class CoreNotifier(Notification):
|
||||||
|
|
||||||
m_lock = threading.RLock()
|
m_lock = threading.Lock()
|
||||||
messages = []
|
messages = []
|
||||||
listeners = []
|
listeners = []
|
||||||
|
|
||||||
@@ -124,14 +124,14 @@ class CoreNotifier(Notification):
|
|||||||
|
|
||||||
self.m_lock.acquire()
|
self.m_lock.acquire()
|
||||||
message = {
|
message = {
|
||||||
'id': str(uuid.uuid4()),
|
'message_id': str(uuid.uuid4()),
|
||||||
'time': time.time(),
|
'time': time.time(),
|
||||||
'type': type,
|
'type': type,
|
||||||
'data': data,
|
'data': data,
|
||||||
}
|
}
|
||||||
self.messages.append(message)
|
self.messages.append(message)
|
||||||
|
|
||||||
while True and not self.shuttingDown():
|
while len(self.listeners) > 0 and not self.shuttingDown():
|
||||||
try:
|
try:
|
||||||
listener, last_id = self.listeners.pop()
|
listener, last_id = self.listeners.pop()
|
||||||
listener({
|
listener({
|
||||||
@@ -142,7 +142,6 @@ class CoreNotifier(Notification):
|
|||||||
break
|
break
|
||||||
|
|
||||||
self.m_lock.release()
|
self.m_lock.release()
|
||||||
|
|
||||||
self.cleanMessages()
|
self.cleanMessages()
|
||||||
|
|
||||||
def addListener(self, callback, last_id = None):
|
def addListener(self, callback, last_id = None):
|
||||||
@@ -157,8 +156,9 @@ class CoreNotifier(Notification):
|
|||||||
|
|
||||||
self.listeners.append((callback, last_id))
|
self.listeners.append((callback, last_id))
|
||||||
|
|
||||||
|
|
||||||
def removeListener(self, callback):
|
def removeListener(self, callback):
|
||||||
self.m_lock.acquire()
|
|
||||||
for list_tuple in self.listeners:
|
for list_tuple in self.listeners:
|
||||||
try:
|
try:
|
||||||
listener, last_id = list_tuple
|
listener, last_id = list_tuple
|
||||||
@@ -166,26 +166,28 @@ class CoreNotifier(Notification):
|
|||||||
self.listeners.remove(list_tuple)
|
self.listeners.remove(list_tuple)
|
||||||
except:
|
except:
|
||||||
pass
|
pass
|
||||||
self.m_lock.release()
|
|
||||||
|
|
||||||
def cleanMessages(self):
|
def cleanMessages(self):
|
||||||
|
|
||||||
self.m_lock.acquire()
|
self.m_lock.acquire()
|
||||||
|
|
||||||
for message in self.messages:
|
for message in self.messages:
|
||||||
if message['time'] < (time.time() - 15):
|
if message['time'] < (time.time() - 15):
|
||||||
self.messages.remove(message)
|
self.messages.remove(message)
|
||||||
|
|
||||||
self.m_lock.release()
|
self.m_lock.release()
|
||||||
|
|
||||||
def getMessages(self, last_id):
|
def getMessages(self, last_id):
|
||||||
self.m_lock.acquire()
|
self.m_lock.acquire()
|
||||||
|
|
||||||
recent = []
|
recent = []
|
||||||
index = 0
|
index = 0
|
||||||
for i in xrange(len(self.messages)):
|
for i in xrange(len(self.messages)):
|
||||||
index = len(self.messages) - i - 1
|
index = len(self.messages) - i - 1
|
||||||
if self.messages[index]["id"] == last_id: break
|
if self.messages[index]["message_id"] == last_id: break
|
||||||
recent = self.messages[index + 1:]
|
recent = self.messages[index + 1:]
|
||||||
|
|
||||||
self.m_lock.release()
|
self.m_lock.release()
|
||||||
|
|
||||||
return recent or []
|
return recent or []
|
||||||
|
|
||||||
def listener(self):
|
def listener(self):
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ var NotificationBase = new Class({
|
|||||||
|
|
||||||
// Listener
|
// Listener
|
||||||
App.addEvent('unload', self.stopPoll.bind(self));
|
App.addEvent('unload', self.stopPoll.bind(self));
|
||||||
|
App.addEvent('reload', self.startInterval.bind(self, [true]));
|
||||||
App.addEvent('notification', self.notify.bind(self));
|
App.addEvent('notification', self.notify.bind(self));
|
||||||
|
|
||||||
// Add test buttons to settings page
|
// Add test buttons to settings page
|
||||||
@@ -86,10 +87,13 @@ var NotificationBase = new Class({
|
|||||||
|
|
||||||
},
|
},
|
||||||
|
|
||||||
startInterval: function(){
|
startInterval: function(force){
|
||||||
var self = this;
|
var self = this;
|
||||||
|
|
||||||
if(self.stopped) return;
|
if(self.stopped && !force){
|
||||||
|
self.stopped = false;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
Api.request('notification.listener', {
|
Api.request('notification.listener', {
|
||||||
'data': {'init':true},
|
'data': {'init':true},
|
||||||
@@ -132,7 +136,7 @@ var NotificationBase = new Class({
|
|||||||
App.fireEvent(result.type, result)
|
App.fireEvent(result.type, result)
|
||||||
})
|
})
|
||||||
|
|
||||||
self.last_id = json.result.getLast().id
|
self.last_id = json.result.getLast().message_id
|
||||||
}
|
}
|
||||||
|
|
||||||
// Restart poll
|
// Restart poll
|
||||||
|
|||||||
Reference in New Issue
Block a user