diff --git a/addons/web_im/im.py b/addons/web_im/im.py index ad6529319d3..8fdcf35d2b7 100644 --- a/addons/web_im/im.py +++ b/addons/web_im/im.py @@ -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):