pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Extracted polling facility

Browse Source 
bzr revid: nicolas.vanhoren@openerp.com-20130128181009-900bgukhf1wc0m2h
This commit is contained in:
niv-openerp
2013-01-28 19:10:09 +01:00
parent 4efa8491a0
commit 71fbc3e352
1 changed files with 68 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/web_im/im.py
+68 -48
View File
@@ -31,8 +31,6 @@ import json
_logger = logging.getLogger(__name__)
WATCHER_TIMER = 60
WATCHER_ERROR_DELAY = 10
POLL_TIMER = 30
DISCONNECTION_TIMER = POLL_TIMER + 5
@@ -41,9 +39,63 @@ if openerp.tools.config.options["gevent"]:
import gevent.event
import select
global ImWatcher
WATCHER_TIMER = 60
WATCHER_ERROR_DELAY = 10
class ImWatcher:
class Watcher(object):
def __init__(self, channel_name, db_name):
self.channel_name = channel_name
self.db_name = db_name
gevent.spawn(self.loop)
def loop(self):
_logger.info("Begin watching on channel "+ self.channel_name +" for database " + self.db_name)
stopping = False
while not stopping:
try:
registry = openerp.modules.registry.RegistryManager.get(self.db_name)
with registry.cursor() as c:
conn = c._cnx
try:
c.execute("listen " + self.channel_name + ";")
c.commit();
while not stopping:
if self.check_stop():
stopping = True
break
if select.select([conn], [], [], WATCHER_TIMER) == ([],[],[]):
pass
else:
conn.poll()
while conn.notifies:
message = json.loads(conn.notifies.pop().payload)
self.handle_message(message)
finally:
try:
c.execute("unlisten " + self.channel_name + ";")
c.commit()
except:
pass # can't do anything if that fails
except:
# if something crash, we wait some time then try again
_logger.exception("Exception during watcher activity")
time.sleep(WATCHER_ERROR_DELAY)
del ImWatcher.watchers[self.db_name]
_logger.info("End watching on channel "+ self.channel_name +" for database " + self.db_name)
def handle_message(self, message):
pass
def check_stop(self):
return False
def post_on_channel(cr, channel_name, message):
cr.commit()
cr.execute("notify " + channel_name + ", %s", [json.dumps(message)])
cr.commit()
class ImWatcher(Watcher):
watchers = {}
@staticmethod
@@ -53,53 +105,23 @@ if openerp.tools.config.options["gevent"]:
return ImWatcher.watchers[db_name]
def __init__(self, db_name):
self.db_name = db_name
ImWatcher.watchers[db_name] = self
self.waiting = 0
self.wait_id = 0
self.users = {}
self.users_watch = {}
gevent.spawn(self.loop)
super(ImWatcher, self).__init__("im_channel", db_name)
def loop(self):
_logger.info("Begin watching for instant messaging events for database " + self.db_name)
stopping = False
while not stopping:
try:
registry = openerp.modules.registry.RegistryManager.get(self.db_name)
with registry.cursor() as c:
conn = c._cnx
try:
c.execute("listen im_channel;")
c.commit();
while not stopping:
if self.waiting == 0:
stopping = True
break
if select.select([conn], [], [], WATCHER_TIMER) == ([],[],[]):
pass
else:
conn.poll()
while conn.notifies:
message = json.loads(conn.notifies.pop().payload)
if message["type"] == "message":
for waiter in self.users.get(message["receiver"], {}).values():
waiter.set()
else: #type status
for waiter in self.users_watch.get(message["user"], {}).values():
waiter.set()
finally:
try:
c.execute("unlisten im_channel;")
c.commit()
except:
pass # can't do anything if that fails
except:
# if something crash, we wait some time then try again
_logger.exception("Exception during instant messaging watcher activity")
time.sleep(WATCHER_ERROR_DELAY)
del ImWatcher.watchers[self.db_name]
_logger.info("End watching for instant messaging events for database " + self.db_name)
def handle_message(self, message):
if message["type"] == "message":
for waiter in self.users.get(message["receiver"], {}).values():
waiter.set()
else: #type status
for waiter in self.users_watch.get(message["user"], {}).values():
waiter.set()
def check_stop(self):
return self.waiting == 0
def _get_wait_id(self):
self.wait_id += 1
@@ -186,9 +208,7 @@ class im_message(osv.osv):
def post(self, cr, uid, message, to_user_id, context=None):
self.create(cr, uid, {"message": message, 'from': uid, 'to': to_user_id}, context=context)
cr.commit()
cr.execute("notify im_channel, %s", [json.dumps({'type': 'message', 'receiver': to_user_id})])
cr.commit()
post_on_channel(cr, "im_channel", {'type': 'message', 'receiver': to_user_id})
return False
class im_user(osv.osv):