Merge lp:~unifield-team/unifield-wm/sync-odoo-UF-2433 into lp:unifield-wm/sync

[jftempo]

=== modified file 'sync_client/monitor.py'
--- sync_client/monitor.py	2014-09-29 15:35:02 +0000
+++ sync_client/monitor.py	2014-10-28 09:21:32 +0000
@@ -35,9 +35,21 @@
         self.info = {
             'status' : 'in-progress',
             'data_pull' : 'null',
+            'data_pull_receive' : 'null',
+            'data_pull_execute' : 'null',
+            
             'msg_pull' : 'null',
+            'msg_pull_receive' : 'null',
+            'msg_pull_execute' : 'null',
+            
+            'msg_push' : 'null',
+            'msg_push_create' : 'null',
+            'msg_push_send' : 'null',
+            
             'data_push' : 'null',
-            'msg_push' : 'null',
+            'data_push_create' : 'null',
+            'data_push_send' : 'null',
+            
         }
         self.info.update(defaults)
         self.final_status = 'ok'
@@ -140,6 +152,18 @@
         return [
             (rec.id, "(%d) %s" % (rec.sequence_number, rec.start))
             for rec in self.browse(cr, user, ids, context=context) ]
+        
+    def interrupt(self, cr, uid, ids, context=None):
+        return self.pool.get('sync.client.entity').interrupt_sync(cr, uid, context=context)
+    
+    def _is_syncing(self, cr, uid, ids, name, arg, context=None):
+        res = dict.fromkeys(ids, "not_syncing")
+        is_syncing = self.pool.get('sync.client.entity').is_syncing()
+        for monitor in self.browse(cr, uid, ids, context=context):
+            if monitor.status == 'in-progress' and is_syncing:
+                res[monitor.id] = "syncing"
+        
+        return res
 
     _rec_name = 'start'
 
@@ -148,11 +172,24 @@
         'start' : fields.datetime("Start Date", readonly=True, required=True),
         'end' : fields.datetime("End Date", readonly=True),
         'data_pull' : fields.selection(status_dict.items(), string="Data Pull", readonly=True),
+        'data_pull_receive' : fields.selection(status_dict.items(), string="DP receive", readonly=True),
+        'data_pull_execute' : fields.selection(status_dict.items(), string="DP execute", readonly=True),
+        
         'msg_pull' : fields.selection(status_dict.items(), string="Msg Pull", readonly=True),
+        'msg_pull_receive' : fields.selection(status_dict.items(), string="MP receive", readonly=True),
+        'msg_pull_execute' : fields.selection(status_dict.items(), string="Msg execute", readonly=True),
+        
         'data_push' : fields.selection(status_dict.items(), string="Data Push", readonly=True),
+        'data_push_create' : fields.selection(status_dict.items(), string="DP create", readonly=True),
+        'data_push_send' : fields.selection(status_dict.items(), string="DP send", readonly=True),
+        
         'msg_push' : fields.selection(status_dict.items(), string="Msg Push", readonly=True),
+        'msg_push_create' : fields.selection(status_dict.items(), string="MP Create", readonly=True),
+        'msg_push_send' : fields.selection(status_dict.items(), string="MP Send", readonly=True),
         'status' : fields.selection(status_dict.items(), string="Status", readonly=True),
         'error' : fields.text("Messages", readonly=True),
+        'state' : fields.function(_is_syncing, method=True, type='selection', string="Is Syncing", selection=[('syncing', 'Syncing'), ('not_syncing', 'Not Syncing')]),
+      
     }
 
     _defaults = {

=== modified file 'sync_client/monitor_view.xml'
--- sync_client/monitor_view.xml	2014-09-05 08:41:11 +0000
+++ sync_client/monitor_view.xml	2014-10-28 09:21:32 +0000
@@ -48,10 +48,20 @@
                     <field name="start"/>
                     <field name="end"/>
                     <field name="data_pull"/>
+                    <field name="data_pull_receive"/>
+                    <field name="data_pull_execute"/>
                     <field name="msg_pull"/>
+                    <field name="msg_pull_receive"/>
+                    <field name="msg_pull_execute"/>
                     <field name="data_push"/>
+                    <field name="data_push_create" />
+                    <field name="data_push_send" />
                     <field name="msg_push"/>
-                    <field name="status"/>
+                    <field name="msg_push_create"/>
+                    <field name="msg_push_send"/>
+                    <field name="status" />
+                    <field name="state"  />
+                    <button name="interrupt" type="object" string="Interrupt"  icon="gtk-cancel"  states="syncing" />
                     <field name="error"/>
                 </tree>
             </field>

=== modified file 'sync_client/rpc.py'
--- sync_client/rpc.py	2014-09-04 13:12:41 +0000
+++ sync_client/rpc.py	2014-10-28 09:21:32 +0000
@@ -30,7 +30,6 @@
 
 #logging.config.fileConfig('logging.cfg')
 
-TIMEOUT = 3600
 GZIP_MAGIC = '\x78\xda' # magic when max compression used
 NB_RETRY = 10
 
@@ -70,13 +69,14 @@
 
     _logger = logging.getLogger('connector')
 
-    def __init__(self, hostname, port):
+    def __init__(self, hostname, port, timeout):
         """
         :param hostname: Host name of the server
         :param port: Port for the connection to the server
         """
         self.hostname = hostname
         self.port = port
+        self.timeout = timeout
 
 class XmlRPCConnector(Connector):
     """
@@ -84,16 +84,35 @@
     """
     PROTOCOL = 'xmlrpc'
 
-    def __init__(self, hostname, port=8069):
-        Connector.__init__(self, hostname, port)
+    def __init__(self, hostname, port=8069, timeout=10.0, retry=0):
+        Connector.__init__(self, hostname, port, timeout=timeout)
         self._logger = logging.getLogger('connector.xmlrpc')
         self.url = 'http://%s:%s/xmlrpc' % (self.hostname, self.port)
+        self.retry = retry
 
     def send(self, service_name, method, *args):
         url = '%s/%s' % (self.url, service_name)
-        timeout_transport = TimeoutTransport(timeout=TIMEOUT)
-        service = xmlrpclib.ServerProxy(url, allow_none=1, transport=timeout_transport)
-        return getattr(service, method)(*args)
+        transport = TimeoutTransport(timeout=self.timeout)
+        service = xmlrpclib.ServerProxy(url, allow_none=1, transport=transport)
+        return self._send(service, method, *args)
+    
+    def _send(self, service, method, *args):
+        i = 0
+        retry = True
+        while retry:
+            try:
+                retry = False
+                return getattr(service, method)(*args)
+            except Exception, e:
+                error = e
+                if i < self.retry:
+                    print 'retry xml_rpc', i
+                    retry = True
+                    self._logger.debug("retry to connect %s, error : %s" ,i, e)
+                i += 1
+        if error:
+            raise osv.except_osv(_('Error!'), "Unable to proceed for the following reason:\n%s" % (e.faultCode if hasattr(e, 'faultCode') else tools.ustr(e)))
+        
 
 """Modified version of xmlrcpclib.Transport.request (same in Python 2.4, 2.5, 2.6)
    to workaround Python bug http://bugs.python.org/issue1223
@@ -131,14 +150,14 @@
     """
     PROTOCOL = 'xmlrpcs'
 
-    def __init__(self, hostname, port=8070):
-        XmlRPCConnector.__init__(self, hostname, port)
+    def __init__(self, hostname, port=8070, timeout=10.0, retry=0):
+        XmlRPCConnector.__init__(self, hostname, port, timeout=timeout, retry=retry)
         self.url = 'https://%s:%s/xmlrpc' % (self.hostname, self.port)
 
     def send(self, service_name, method, *args):
         url = '%s/%s' % (self.url, service_name)
         service = xmlrpclib.ServerProxy(url, allow_none=1)
-        return getattr(service, method)(*args)
+        return self._send(service, method, *args)
 
 class GzipXmlRPCConnector(XmlRPCConnector):
     """
@@ -148,9 +167,9 @@
 
     def send(self, service_name, method, *args):
         url = '%s/%s' % (self.url, service_name)
-        gzip_transport = GzipTransport(timeout=TIMEOUT)
+        gzip_transport = GzipTransport(timeout=self.timeout)
         service = xmlrpclib.ServerProxy(url, allow_none=1, transport=gzip_transport)
-        return getattr(service, method)(*args)
+        return self._send(service, method, *args)
 
 class GzipXmlRPCSConnector(GzipXmlRPCConnector):
     PROTOCOL = 'gzipxmlrpcs'
@@ -172,13 +191,13 @@
         self.args = (faultCode, faultString)
 
 class NetRPC:
-    def __init__(self, sock=None, is_gzip=False):
+    def __init__(self, sock=None, is_gzip=False, timeout=10.0):
         if sock is None:
             self.sock = socket.socket(
             socket.AF_INET, socket.SOCK_STREAM)
         else:
             self.sock = sock
-        self.sock.settimeout(TIMEOUT)
+        self.sock.settimeout(timeout)
         self.is_gzip = is_gzip
         self._logger = logging.getLogger('netrpc')
 
@@ -202,7 +221,7 @@
             raw_size = len(msg)
             msg = zlib.compress(msg, zlib.Z_BEST_COMPRESSION)
             gzipped_size = len(msg)
-            saving = 100*(float(raw_size-gzipped_size))/gzipped_size if gzipped_size else 0
+            #saving = 100*(float(raw_size-gzipped_size))/gzipped_size if gzipped_size else 0
             #self._logger.debug('payload size: raw %s, gzipped %s, saving %.2f%%', raw_size, gzipped_size, saving)
         size = len(msg)
         self.sock.send('%8d' % size)
@@ -237,7 +256,7 @@
             gzipped_size = len(msg)
             msg = zlib.decompress(msg)
             raw_size = len(msg)
-            saving = 100*(float(raw_size-gzipped_size))/gzipped_size if gzipped_size else 0
+            #saving = 100*(float(raw_size-gzipped_size))/gzipped_size if gzipped_size else 0
             #self._logger.debug('payload size: raw %s, gzipped %s, saving %.2f%%', raw_size, gzipped_size, saving)
         res = SafeUnpickler.loads(msg)
 
@@ -251,10 +270,11 @@
 class NetRPCConnector(Connector):
     PROTOCOL = 'netrpc'
 
-    def __init__(self, hostname, port=8070, is_gzip=False):
-        Connector.__init__(self, hostname, port)
+    def __init__(self, hostname, port=8070, is_gzip=False, timeout=10.0, retry=10):
+        Connector.__init__(self, hostname, port, timeout=timeout)
         self._logger = logging.getLogger('connector.netrpc')
         self.is_gzip = is_gzip
+        self.retry = retry
 
     def send(self, service_name, method, *args):
         i = 0
@@ -264,13 +284,13 @@
         while retry:
             try:
                 retry = False
-                socket = NetRPC(is_gzip=self.is_gzip)
+                socket = NetRPC(is_gzip=self.is_gzip, timeout=self.timeout)
                 socket.connect(self.hostname, self.port)
                 socket.mysend((service_name, method, )+args)
                 result = socket.myreceive()
             except Exception, e:
                 error = e
-                if i < NB_RETRY:
+                if i < self.retry:
                     retry = True
                     self._logger.debug("retry to connect %s, error : %s" ,i, e)
                 i += 1

=== modified file 'sync_client/sync_client.py'
--- sync_client/sync_client.py	2014-10-16 09:10:47 +0000
+++ sync_client/sync_client.py	2014-10-28 09:21:32 +0000
@@ -90,7 +90,44 @@
             pass
         finally:
             cr.close()
-
+            
+def sync_subprocess(step='status', defaults_logger={}):
+    def decorator(fn):
+
+        @functools.wraps(fn)
+        def wrapper(self, cr, uid, *args, **kwargs):
+            context = kwargs['context'] = kwargs.get('context') is not None and dict(kwargs.get('context', {})) or {}
+            logger = context.get('logger')
+            logger.switch(step, 'in-progress')
+            logger.write()
+            try:
+                res = fn(self, self.sync_cursor, uid, *args, **kwargs)
+            except osv.except_osv, e:
+                logger.switch(step, 'failed')
+                raise
+            except BaseException, e:
+                # Handle aborting of synchronization
+                if isinstance(e, OperationalError) and e.message == 'Unable to use the cursor after having closed it':
+                    logger.switch(step, 'aborted')
+                    self.sync_cursor = None
+                    raise
+                logger.switch(step, 'failed')
+                error = "%s: %s" % (e.__class__.__name__, getattr(e, 'message', tools.ustr(e)))
+                self._logger.exception('Error in sync_process at step %s' % step)
+                logger.append(error, step)
+                raise
+            else:
+                logger.switch(step, 'ok')
+                if isinstance(res, (str, unicode)) and res:
+                    logger.append(res, step)
+            finally:
+                # gotcha!
+                logger.write()
+            return res
+        return wrapper
+    return decorator
+
+    
 def sync_process(step='status', need_connection=True, defaults_logger={}):
     is_step = not (step == 'status')
 
@@ -372,10 +409,10 @@
 
 
         return True
-
+    
+    @sync_subprocess('data_push_create')
     def create_update(self, cr, uid, context=None):
         context = context or {}
-        logger = context.get('logger')
         updates = self.pool.get(context.get('update_to_send_model', 'sync.client.update_to_send'))
 
         def set_rules(identifier):
@@ -403,6 +440,7 @@
         return updates_count
         #state init => update_send
 
+    @sync_subprocess('data_push_send')
     def send_update(self, cr, uid, context=None):
         context = context or {}
         logger = context.get('logger')
@@ -508,13 +546,13 @@
 
         return True
 
+    @sync_subprocess('data_pull_receive')
     def retrieve_update(self, cr, uid, max_packet_size, recover=False, context=None):
         context = context or {}
         logger = context.get('logger')
         updates = self.pool.get(context.get('update_received_model', 'sync.client.update_received'))
 
         entity = self.get_entity(cr, uid, context)
-        last = False
         last_seq = entity.update_last
         max_seq = entity.max_update
         offset = entity.update_offset
@@ -546,6 +584,7 @@
 
         return updates_count
 
+    @sync_subprocess('data_pull_execute')
     def execute_updates(self, cr, uid, context=None):
         context = context or {}
         logger = context.get('logger')
@@ -637,6 +676,7 @@
 
         return True
 
+    @sync_subprocess('msg_push_create')
     def create_message(self, cr, uid, context=None):
         context = context or {}
         messages = self.pool.get(context.get('message_to_send_model', 'sync.client.message_to_send'))
@@ -656,7 +696,8 @@
             messages_count += messages.create_from_rule(cr, uid, rule, None, context=context)
 
         return messages_count
-
+    
+    @sync_subprocess('msg_push_send')
     def send_message(self, cr, uid, context=None):
         context = context or {}
         logger = context.get('logger')
@@ -696,8 +737,11 @@
     @sync_process('msg_pull')
     def pull_message(self, cr, uid, recover=False, context=None):
         context = context or {}
+<<<<<<< TREE
         logger = context.get('logger')
         proxy = self.pool.get("sync.client.sync_server_connection").get_connection(cr, uid, "sync.server.sync_manager")
+=======
+>>>>>>> MERGE-SOURCE
 
         entity = self.get_entity(cr, uid, context=context)
         if recover:
@@ -712,7 +756,8 @@
         msg_count = self.execute_message(cr, uid, context=context)
         self._logger.info("Pull message :: Number of messages pulled: %s" % msg_count)
         return True
-
+    
+    @sync_subprocess('msg_pull_receive')
     def get_message(self, cr, uid, context=None):
         context = context or {}
         logger = context.get('logger')
@@ -746,6 +791,7 @@
 
         return messages_count
 
+    @sync_subprocess('msg_pull_execute')
     def execute_message(self, cr, uid, context=None):
         context = context or {}
         logger = context.get('logger')
@@ -894,6 +940,15 @@
                 % (_(monitor.status_dict[last_log[0]]), last_log[1])
 
         return "Connected"
+    
+    def interrupt_sync(self, cr, uid, context=None):
+        if self.is_syncing():
+            try:
+                self._renew_sync_lock()
+            except StandardError:
+                return False
+            self.sync_cursor.close()
+        return True
 
 Entity()
 
@@ -952,6 +1007,9 @@
         'password': fields.function(_get_password, fnct_inv=_set_password, string='Password', type='char', method=True, store=False),
         'state' : fields.function(_get_state, method=True, string='State', type="char", readonly=True, store=False),
         'max_size' : fields.integer("Max Packet Size"),
+        'timeout' : fields.float("Timeout"),
+        'netrpc_retry' : fields.integer("NetRPC retry"),
+        'xmlrpc_retry' : fields.integer("XmlRPC retry"),
     }
 
     _defaults = {
@@ -962,6 +1020,9 @@
         'login' : 'admin',
         'max_size' : 500,
         'database' : 'SYNC_SERVER',
+        'timeout' : 10.0,
+        'netrpc_retry' : 0,
+        'xmlrpc_retry' : 0,
     }
 
     def _get_connection_manager(self, cr, uid, context=None):
@@ -972,17 +1033,21 @@
 
     def connector_factory(self, con):
         if con.protocol == 'xmlrpc':
-            connector = rpc.XmlRPCConnector(con.host, con.port)
+            connector = rpc.XmlRPCConnector(con.host, con.port, timeout=con.timeout, retry=con.xmlrpc_retry)
         elif con.protocol == 'gzipxmlrpc':
+<<<<<<< TREE
             connector = rpc.GzipXmlRPCConnector(con.host, con.port)
         elif con.protocol == 'gzipxmlrpcs':
             connector = rpc.GzipXmlRPCSConnector(con.host, con.port)
+=======
+            connector = rpc.GzipXmlRPCConnector(con.host, con.port, timeout=con.timeout, retry=con.xmlrpc_retry)
+>>>>>>> MERGE-SOURCE
         elif con.protocol == 'xmlrpcs':
-            connector = rpc.SecuredXmlRPCConnector(con.host, con.port)
+            connector = rpc.SecuredXmlRPCConnector(con.host, con.port, timeout=con.timeout, retry=con.xmlrpc_retry)
         elif con.protocol == 'netrpc':
-            connector = rpc.NetRPCConnector(con.host, con.port)
+            connector = rpc.NetRPCConnector(con.host, con.port, timeout=con.timeout, retry=con.netrpc_retry)
         elif con.protocol == 'netrpc_gzip':
-            connector = rpc.GzipNetRPCConnector(con.host, con.port)
+            connector = rpc.GzipNetRPCConnector(con.host, con.port, timeout=con.timeout, retry=con.netrpc_retry)
         else:
             raise osv.except_osv('Connection Error','Unknown protocol: %s' % con.protocol)
         return connector
@@ -1030,6 +1095,7 @@
         return rpc.Object(cnx, model)
 
     def disconnect(self, cr, uid, context=None):
+<<<<<<< TREE
         con = self._get_connection_manager(cr, uid, context=context)
         sync_args = {
             'client_name': cr.dbname,
@@ -1043,6 +1109,10 @@
                 self._logger.warning('Error during the disconnection of client \'%(client_name)s\'' % sync_args)
                 return False
             entity.sync_cursor.close()
+=======
+        if not self.pool.get('sync.client.entity').interrupt_sync(cr, uid, context=context):
+            return False
+>>>>>>> MERGE-SOURCE
         self._uid = False
         self._logger.info('Client \'%(client_name)s\' succesfully disconnected from the sync. server \'%(server_name)s\'' % sync_args)
         return True

=== modified file 'sync_client/sync_client_view.xml'
--- sync_client/sync_client_view.xml	2014-09-08 08:48:22 +0000
+++ sync_client/sync_client_view.xml	2014-10-28 09:21:32 +0000
@@ -32,6 +32,10 @@
                     <field name="login" required="1"/>
                     <field name="password" required="1" password="1"/>
                     <field name="max_size" required="1"/>
+                    <field name="timeout" required="1"/>
+                    <label string="" colspan="2" />
+                    <field name="netrpc_retry" required="1" />
+                    <field name="xmlrpc_retry" required="1" />
                     <separator string="Connection state" colspan="4"/>
                     <!-- the following line break the 6.1 web client backport because its a functional field of type integer -->
                     <!-- <field name="uid" invisible="1"/> -->

=== modified file 'sync_client/timeout_transport.py'
--- sync_client/timeout_transport.py	2014-09-04 13:12:41 +0000
+++ sync_client/timeout_transport.py	2014-10-28 09:21:32 +0000
@@ -1,5 +1,6 @@
 import httplib
 import xmlrpclib
+import sys
 
 class TimeoutHTTPConnection(httplib.HTTPConnection):
 
@@ -7,6 +8,9 @@
         httplib.HTTPConnection.connect(self)
         if self.timeout is not None:
             self.sock.settimeout(self.timeout)
+            
+    def set_timeout(self, timeout):
+        self.timeout = timeout
 
 class TimeoutHTTP(httplib.HTTP):
 
@@ -22,10 +26,13 @@
         self.timeout = timeout
 
     def make_connection(self, host):
-        # TODO: check make_connection for python > 2.6
-        conn = TimeoutHTTP(host)
-        conn.set_timeout(self.timeout)
-        return conn
+        chost, self._extra_headers, _ = self.get_host_info(host)
+        if sys.version_info < (2,7):
+            self._connection = host, TimeoutHTTP(host)
+        else:
+            self._connection = host, TimeoutHTTPConnection(chost)
+        self._connection[1].set_timeout(self.timeout)
+        return self._connection[1]
 
 class TimeoutHTTPSConnection(httplib.HTTPSConnection):
 

=== modified file 'sync_server/sync_server.py'
--- sync_server/sync_server.py	2014-09-19 08:10:11 +0000
+++ sync_server/sync_server.py	2014-10-28 09:21:32 +0000
@@ -346,7 +346,7 @@
         return (True, {
             'name': entity.name,
             'identifier': entity.identifier,
-            'parent': entity.parent_id.name,
+            'parent': entity.parent_id.name or '',
             'email': entity.email,
             'entity_status': entity.state,
             'group': ', '.join([group.name for group in entity.group_ids]),