pysyncobj/transport.py
def _onIncomingMessageReceived(self, conn, message):
        """
        Callback for initial messages on incoming connections. Handles encryption, utility messages, and association of the connection with a Node.
        Once this initial setup is done, the relevant connected callback is executed, and further messages are deferred to the onMessageReceived callback.

        :param conn: connection object
        :type conn: TcpConnection
        :param message: received message
        :type message: any
        """

        if self._syncObj.encryptor and not conn.sendRandKey:
            conn.sendRandKey = message
            conn.recvRandKey = os.urandom(32)
            conn.send(conn.recvRandKey)
            return

        # Utility messages
        if isinstance(message, list):
            done = False
            try:
                if message[0] == 'status':
                    conn.send(self._syncObj.getStatus())
                    done = True
                elif message[0] == 'add':
                    self._syncObj.addNodeToCluster(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'ADD', arg = message[1]))
                    done = True
                elif message[0] == 'remove':
                    if message[1] == self._selfNode.address:
                        conn.send('FAIL REMOVE ' + message[1])
                    else:
                        self._syncObj.removeNodeFromCluster(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'REMOVE', arg = message[1]))
                    done = True
                elif message[0] == 'set_version':
                    self._syncObj.setCodeVersion(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'SET_VERSION', arg = str(message[1])))
                    done = True
            except Exception as e:
                conn.send(str(e))
                done = True
            if done:
                return

        # At this point, message should be either a node ID (i.e. address) or 'readonly'
        node = self._nodeAddrToNode[message] if message in self._nodeAddrToNode else None

        if node is None and message != 'readonly':
            conn.disconnect()
            self._unknownConnections.discard(conn)
            return

        readonly = node is None
        if readonly:
            nodeId = str(self._readonlyNodesCounter)
            node = Node(nodeId)
            self._readonlyNodes.add(node)
            self._readonlyNodesCounter += 1

        self._unknownConnections.discard(conn)
        self._connections[node] = conn
        conn.setOnMessageReceivedCallback(functools.partial(self._onMessageReceived, node))
        if not readonly:
            self._onNodeConnected(node)
        else:
            self._onReadonlyNodeConnected(node)
def _onIncomingMessageReceived(self, conn, message):
        if self._syncObj.encryptor and not conn.sendRandKey:
            conn.sendRandKey = message
            conn.recvRandKey = os.urandom(32)
            conn.send(conn.recvRandKey)
            return

        # Utility messages
        if isinstance(message, list):
            done = False
            try:
                if message[0] == 'status':
                    conn.send(self._syncObj.getStatus())
                    done = True
                elif message[0] == 'add':
                    self._syncObj.addNodeToCluster(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'ADD', arg = message[1]))
                    done = True
                elif message[0] == 'remove':
                    if message[1] == self._selfNode.address:
                        conn.send('FAIL REMOVE ' + message[1])
                    else:
                        self._syncObj.removeNodeFromCluster(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'REMOVE', arg = message[1]))
                    done = True
                elif message[0] == 'set_version':
                    self._syncObj.setCodeVersion(message[1], callback = functools.partial(self._utilityCallback, conn = conn, cmd = 'SET_VERSION', arg = str(message[1])))
                    done = True
            except Exception as e:
                conn.send(str(e))
                done = True
            if done:
                return

        # At this point, message should be either a node ID (i.e. address) or 'readonly'
        node = self._nodeAddrToNode[message] if message in self._nodeAddrToNode else None

        if node is None and message != 'readonly':
            conn.disconnect()
            self._unknownConnections.discard(conn)
            return

        readonly = node is None
        if readonly:
            nodeId = str(self._readonlyNodesCounter)
            node = Node(nodeId)
            self._readonlyNodes.add(node)
            self._readonlyNodesCounter += 1

        self._unknownConnections.discard(conn)
        self._connections[node] = conn
        conn.setOnMessageReceivedCallback(functools.partial(self._onMessageReceived, node))
        if not readonly:
            self._onNodeConnected(node)
        else:
            self._onReadonlyNodeConnected(node)
def _onIncomingMessageReceived ( self , conn , message ) : if self . _syncObj . encryptor and not conn . sendRandKey : conn . sendRandKey = message conn . recvRandKey = os . urandom ( 32 ) conn . send ( conn . recvRandKey ) return # Utility messages if isinstance ( message , list ) : done = False try : if message [ 0 ] == 'status' : conn . send ( self . _syncObj . getStatus ( ) ) done = True elif message [ 0 ] == 'add' : self . _syncObj . addNodeToCluster ( message [ 1 ] , callback = functools . partial ( self . _utilityCallback , conn = conn , cmd = 'ADD' , arg = message [ 1 ] ) ) done = True elif message [ 0 ] == 'remove' : if message [ 1 ] == self . _selfNode . address : conn . send ( 'FAIL REMOVE ' + message [ 1 ] ) else : self . _syncObj . removeNodeFromCluster ( message [ 1 ] , callback = functools . partial ( self . _utilityCallback , conn = conn , cmd = 'REMOVE' , arg = message [ 1 ] ) ) done = True elif message [ 0 ] == 'set_version' : self . _syncObj . setCodeVersion ( message [ 1 ] , callback = functools . partial ( self . _utilityCallback , conn = conn , cmd = 'SET_VERSION' , arg = str ( message [ 1 ] ) ) ) done = True except Exception as e : conn . send ( str ( e ) ) done = True if done : return # At this point, message should be either a node ID (i.e. address) or 'readonly' node = self . _nodeAddrToNode [ message ] if message in self . _nodeAddrToNode else None if node is None and message != 'readonly' : conn . disconnect ( ) self . _unknownConnections . discard ( conn ) return readonly = node is None if readonly : nodeId = str ( self . _readonlyNodesCounter ) node = Node ( nodeId ) self . _readonlyNodes . add ( node ) self . _readonlyNodesCounter += 1 self . _unknownConnections . discard ( conn ) self . _connections [ node ] = conn conn . setOnMessageReceivedCallback ( functools . partial ( self . _onMessageReceived , node ) ) if not readonly : self . _onNodeConnected ( node ) else : self . _onReadonlyNodeConnected ( node )
Callback for initial messages on incoming connections. Handles encryption, utility messages, and association of the connection with a Node.
        Once this initial setup is done, the relevant connected callback is executed, and further messages are deferred to the onMessageReceived callback.

        :param conn: connection object
        :type conn: TcpConnection
        :param message: received message
        :type message: any
Callback for initial messages on incoming connections . Handles encryption utility messages and association of the connection with a Node . Once this initial setup is done the relevant connected callback is executed and further messages are deferred to the onMessageReceived callback .