addons/im/watcher.py

   1
   2 import openerp
   3 import openerp.tools.config
   4 import openerp.modules.registry
   5 from openerp.tools.misc import DEFAULT_SERVER_DATETIME_FORMAT
   6 import datetime
   7 from openerp.osv import osv, fields
   8 import time
   9 import logging
  10 import json
  11 import select
  12 import gevent
  13 import gevent.event
  14 from openerp.addons.im.im import *
  15
  16 _logger = logging.getLogger(__name__)
  17
  18 class ImWatcher(object):
  19     watchers = {}
  20
  21     @staticmethod
  22     def get_watcher(db_name):
  23         if not ImWatcher.watchers.get(db_name):
  24             ImWatcher(db_name)
  25         return ImWatcher.watchers[db_name]
  26
  27     def __init__(self, db_name):
  28         self.db_name = db_name
  29         ImWatcher.watchers[db_name] = self
  30         self.waiting = 0
  31         self.wait_id = 0
  32         self.users = {}
  33         self.users_watch = {}
  34         gevent.spawn(self.loop)
  35
  36     def loop(self):
  37         _logger.info("Begin watching on channel im_channel for database " + self.db_name)
  38         stop = False
  39         while not stop:
  40             try:
  41                 registry = openerp.modules.registry.RegistryManager.get(self.db_name)
  42                 with registry.cursor() as cr:
  43                     listen_channel(cr, "im_channel", self.handle_message, self.check_stop)
  44                     stop = True
  45             except:
  46                 # if something crash, we wait some time then try again
  47                 _logger.exception("Exception during watcher activity")
  48                 time.sleep(WATCHER_ERROR_DELAY)
  49         _logger.info("End watching on channel im_channel for database " + self.db_name)
  50         del ImWatcher.watchers[self.db_name]
  51
  52     def handle_message(self, message):
  53         if message["type"] == "message":
  54             for receiver in message["receivers"]:
  55                 for waiter in self.users.get(receiver, {}).values():
  56                     waiter.set()
  57         else: #type status
  58             for waiter in self.users_watch.get(message["user"], {}).values():
  59                 waiter.set()
  60
  61     def check_stop(self):
  62         return self.waiting == 0
  63
  64     def _get_wait_id(self):
  65         self.wait_id += 1
  66         return self.wait_id
  67
  68     def stop(self, user_id, watch_users, timeout=None):
  69         wait_id = self._get_wait_id()
  70         event = gevent.event.Event()
  71         self.waiting += 1
  72         self.users.setdefault(user_id, {})[wait_id] = event
  73         for watch in watch_users:
  74             self.users_watch.setdefault(watch, {})[wait_id] = event
  75         try:
  76             event.wait(timeout)
  77         finally:
  78             for watch in watch_users:
  79                 del self.users_watch[watch][wait_id]
  80                 if len(self.users_watch[watch]) == 0:
  81                     del self.users_watch[watch]
  82             del self.users[user_id][wait_id]
  83             if len(self.users[user_id]) == 0:
  84                 del self.users[user_id]
  85             self.waiting -= 1