feat(rfc): Now handle late join

net_components.py

@@ -9,6 +9,7 @@ import struct
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
 
+
 class RCFTranslation():
     def get(self,data):
         """
@@ -72,18 +73,20 @@ class RCFMessage(object):
         # this seems weird to check, but it's what the C example does
         # this currently erasing old value
         if self.key is not None and self.body is not None:
-            if self.key in dikt:
             dikt[self.key] = self
                 
 
     def send(self, socket):
         """Send key-value message to socket; any empty frames are sent as such."""
-        key = '' if self.key is None else self.key.encode()
+        key = ''.encode() if self.key is None else self.key.encode()
-        mtype = '' if self.mtype is None else self.mtype.encode()
+        mtype = ''.encode() if self.mtype is None else self.mtype.encode()
-        body = '' if self.body is None else umsgpack.packb(self.body)
+        body = ''.encode() if self.body is None else umsgpack.packb(self.body)
-        factory = '' if self.factory is None else umsgpack.packb(self.factory)
+        id = ''.encode()  if self.id is None else self.id
 
-        socket.send_multipart([key, self.id, mtype, body,factory])
+        try:
+            socket.send_multipart([key, id, mtype, body])
+        except:
+            logger.info("Fail to send {}".format(key))
 
     @classmethod
     def recv(cls, socket):
@@ -112,7 +115,7 @@ class RCFMessage(object):
 
 
 class Client():
-    def __init__(self, context=zmq.Context(), id="default", recv_callback=None,is_admin=False):
+    def __init__(self, context=zmq.Context(), id="default", on_recv=None,on_post_init=None,is_admin=False):
         self.is_admin = is_admin
 
         #0MQ vars
@@ -122,8 +125,10 @@ class Client():
         self.poller = None
 
         self.id = id.encode()
-        self.recv_callback = recv_callback
+        self.on_recv = on_recv
+        self.on_post_init = on_post_init
         self.bind_ports()
+
         # Main client loop registration
         self.task = asyncio.ensure_future(self.main())
 
@@ -141,7 +146,7 @@ class Client():
         # request socket: send request/message over all peers throught the server
         self.req_sock = self.context.socket(zmq.DEALER)
         self.req_sock.setsockopt(zmq.IDENTITY, self.id)
-        self.req_sock.setsockopt(zmq.SNDHWM, 60)
+        # self.req_sock.setsockopt(zmq.SNDHWM, 60)
         self.req_sock.linger = 0
         self.req_sock.connect("tcp://localhost:5556")
 
@@ -156,17 +161,17 @@ class Client():
         self.poller = zmq.Poller()
         self.poller.register(self.pull_sock, zmq.POLLIN)
 
-    async def main(self):
+        time.sleep(0.5)
-        logger.info("{} client launched".format(id))
 
-        # Late join
+    async def main(self):
+        logger.info("{} client syncing".format(id))
+
+        # Late join mecanism
         logger.info("{} send snapshot request".format(id))
-        self.req_snapshot.send(b"SNAPSHOT_REQUEST")
+        self.req_sock.send(b"SNAPSHOT_REQUEST")
         while True:
             try:
-                rcfmsg_snapshot = RCFMessage.recv(snapshot)
+                rcfmsg_snapshot = RCFMessage.recv(self.req_sock)
-            except:
-                return 
 
                 if rcfmsg_snapshot.key == "SNAPSHOT_END":
                     logger.info("snapshot complete")
@@ -174,8 +179,14 @@ class Client():
                 else:
                     logger.info("received : {}".format(rcfmsg_snapshot.key))
                     rcfmsg_snapshot.store(self.property_map)
+            except:
+                await asyncio.sleep(0.001)
         
-        # Prepare our context and publisher socket
+        for f in self.on_post_init:
+                f()
+        logger.info("{} client running".format(id))
+
+        # Main loop
         while True:
             # TODO: find a better way
             socks = dict(self.poller.poll(1))
@@ -185,7 +196,7 @@ class Client():
 
                 rcfmsg.store(self.property_map)
 
-                for f in self.recv_callback:
+                for f in self.on_recv:
                     f(rcfmsg)
             else:
                 await asyncio.sleep(0.001)
@@ -235,6 +246,7 @@ class Server():
 
         # Update request
         self.request_sock = self.context.socket(zmq.ROUTER)
+        self.request_sock.setsockopt(zmq.IDENTITY, b'SERVER')
         self.request_sock.setsockopt(zmq.RCVHWM, 60)
         self.request_sock.bind("tcp://*:5556")
 
@@ -255,11 +267,13 @@ class Server():
             # TODO: Listener on anoter process linked with PAIR/PAIR ?
             socks = dict(self.poller.poll(1))
 
+            # Snapshot system for late join
             if self.request_sock in socks:
-                msg = self.request_sock.recv_multipart()
+                msg = self.request_sock.recv_multipart(zmq.DONTWAIT)
             
                 identity = msg[0]
                 request = msg[1]
+                print("reveived snapshot request from {}".format(identity.decode()))
                 if request == b"SNAPSHOT_REQUEST":
                     pass
                 else:
@@ -271,17 +285,15 @@ class Server():
                     self.request_sock.send(identity,zmq.SNDMORE)
                     v.send(self.request_sock)
                 
-                logger.info("done".format(k,identity))
+                msg_end_snapshot = RCFMessage(key="SNAPSHOT_END",id=identity)
-                
-                msg_end_snapshot = RCFMessage(key="SNAPSHOT_END")
                 self.request_sock.send(identity,zmq.SNDMORE)
                 msg_end_snapshot.send(self.request_sock)
-
+                logger.info("done")
             elif self.collector_sock in socks:
-                msg = self.collector_sock.recv_multipart(zmq.DONTWAIT)
+                msg = RCFMessage.recv(self.collector_sock)
                 # Update all clients
-
+                msg.store(self.property_map)
-                self.pub_sock.send_multipart(msg)
+                msg.send(self.pub_sock)
             else:
                 await asyncio.sleep(0.016)
 

net_operators.py

@@ -3,6 +3,8 @@ from . import net_components
 import time
 import logging
 import mathutils
+import random
+import string
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +23,12 @@ VECTOR_TYPES = (
     'Vector'
 )
 
+
+def randomStringDigits(stringLength=6):
+    """Generate a random string of letters and digits """
+    lettersAndDigits = string.ascii_letters + string.digits
+    return ''.join(random.choice(lettersAndDigits) for i in range(stringLength))
+
 class VectorTypeTranslation(net_components.RCFTranslation):
     def set(self,data):
         """
@@ -35,7 +43,6 @@ class VectorTypeTranslation(net_components.RCFTranslation):
         """
         return mathutils.Vector((data[0],data[1],data[2]))
 
-
 def match_supported_types(value):
     type_factory = None
 
@@ -147,7 +154,7 @@ def observer():
 
 
 # CLIENT-SERVER
-def refresh_window(msg):
+def refresh_window():
     import bpy
 
     bpy.ops.wm.redraw_timer(type='DRAW_WIN_SWAP', iterations=1)
@@ -178,6 +185,10 @@ def array2vector(a):
     logger.debug(mathutils.Vector((a[0],a[1],a[2])))
     return mathutils.Vector((a[0],a[1],a[2]))
 
+
+recv_callbacks=[patch_scene]
+post_init_callbacks=[refresh_window]
+
 class session_join(bpy.types.Operator):
     bl_idname = "session.join"
     bl_label = "join"
@@ -191,10 +202,15 @@ class session_join(bpy.types.Operator):
     def execute(self, context):
         global client
 
-        username = str(context.scene.session_settings.username)
+        net_settings = context.scene.session_settings
-        callbacks=[patch_scene]
         
-        client = net_components.Client(id=username,recv_callback=callbacks)
+        if net_settings.username == "DefaultUser":
+            net_settings.username = "{}_{}".format(net_settings.username,randomStringDigits())
+
+        username = str(context.scene.session_settings.username)
+        
+
+        client = net_components.Client(id=username,on_recv=recv_callbacks,on_post_init=post_init_callbacks)
         time.sleep(1)
 
         bpy.ops.asyncio.loop()
@@ -202,7 +218,6 @@ class session_join(bpy.types.Operator):
 
         return {"FINISHED"}
 
-
 class session_send(bpy.types.Operator):
     bl_idname = "session.send"
     bl_label = "send"
@@ -242,13 +257,19 @@ class session_create(bpy.types.Operator):
         global server
         global client
         
-        username = str(context.scene.session_settings.username)
+        # net_settings = context.scene.session_settings
-        callbacks=[patch_scene]
+        
+        # if net_settings.username == "DefaultUser":
+        #     net_settings.username = "{}_{}".format(net_settings.username,randomStringDigits())
+
+        # username = str(context.scene.session_settings.username)
         
         server = net_components.Server()
-        client = net_components.Client(id=username,recv_callback=callbacks)
+        time.sleep(0.1)
+
+        bpy.ops.session.join()
+        # client = net_components.Client(id=username,on_recv=recv_callbacks,on_post_init=post_init_callbacks)
 
-        # time.sleep(0.1)
     
         bpy.ops.asyncio.loop()
         bpy.app.timers.register(observer)
@@ -306,6 +327,7 @@ def register():
         register_class(cls)
 
     bpy.types.Scene.session_settings = bpy.props.PointerProperty(type=session_settings)
+    
     # bpy.app.handlers.depsgraph_update_post.append(observer)
     
 def unregister():