connectordb/_websocket.py
def __on_message(self, ws, msg):
        """This function is called whenever there is a message received from the server"""
        msg = json.loads(msg)
        logging.debug("ConnectorDB:WS: Msg '%s'", msg["stream"])

        # Build the subcription key
        stream_key = msg["stream"] + ":"
        if "transform" in msg:
            stream_key += msg["transform"]

        self.subscription_lock.acquire()
        if stream_key in self.subscriptions:
            subscription_function = self.subscriptions[stream_key]
            self.subscription_lock.release()

            fresult = subscription_function(msg["stream"], msg["data"])

            if fresult is True:
                # This is a special result - if the subscription function of a downlink returns True,
                # then the datapoint is acknowledged automatically (ie, reinserted in non-downlink stream)
                fresult = msg["data"]

            if fresult is not False and fresult is not None and msg["stream"].endswith(
                    "/downlink") and msg["stream"].count("/") == 3:
                # If the above conditions are true, it means that the datapoints were from a downlink,
                # and the subscriber function chooses to acknowledge them, so we reinsert them.
                self.insert(msg["stream"][:-9], fresult)
        else:
            self.subscription_lock.release()
            logging.warn(
                "ConnectorDB:WS: Msg '%s' not subscribed! Subscriptions: %s",
                msg["stream"], list(self.subscriptions.keys()))
def __on_message(self, ws, msg):
        msg = json.loads(msg)
        logging.debug("ConnectorDB:WS: Msg '%s'", msg["stream"])

        # Build the subcription key
        stream_key = msg["stream"] + ":"
        if "transform" in msg:
            stream_key += msg["transform"]

        self.subscription_lock.acquire()
        if stream_key in self.subscriptions:
            subscription_function = self.subscriptions[stream_key]
            self.subscription_lock.release()

            fresult = subscription_function(msg["stream"], msg["data"])

            if fresult is True:
                # This is a special result - if the subscription function of a downlink returns True,
                # then the datapoint is acknowledged automatically (ie, reinserted in non-downlink stream)
                fresult = msg["data"]

            if fresult is not False and fresult is not None and msg["stream"].endswith(
                    "/downlink") and msg["stream"].count("/") == 3:
                # If the above conditions are true, it means that the datapoints were from a downlink,
                # and the subscriber function chooses to acknowledge them, so we reinsert them.
                self.insert(msg["stream"][:-9], fresult)
        else:
            self.subscription_lock.release()
            logging.warn(
                "ConnectorDB:WS: Msg '%s' not subscribed! Subscriptions: %s",
                msg["stream"], list(self.subscriptions.keys()))
def __on_message ( self , ws , msg ) : msg = json . loads ( msg ) logging . debug ( "ConnectorDB:WS: Msg '%s'" , msg [ "stream" ] ) # Build the subcription key stream_key = msg [ "stream" ] + ":" if "transform" in msg : stream_key += msg [ "transform" ] self . subscription_lock . acquire ( ) if stream_key in self . subscriptions : subscription_function = self . subscriptions [ stream_key ] self . subscription_lock . release ( ) fresult = subscription_function ( msg [ "stream" ] , msg [ "data" ] ) if fresult is True : # This is a special result - if the subscription function of a downlink returns True, # then the datapoint is acknowledged automatically (ie, reinserted in non-downlink stream) fresult = msg [ "data" ] if fresult is not False and fresult is not None and msg [ "stream" ] . endswith ( "/downlink" ) and msg [ "stream" ] . count ( "/" ) == 3 : # If the above conditions are true, it means that the datapoints were from a downlink, # and the subscriber function chooses to acknowledge them, so we reinsert them. self . insert ( msg [ "stream" ] [ : - 9 ] , fresult ) else : self . subscription_lock . release ( ) logging . warn ( "ConnectorDB:WS: Msg '%s' not subscribed! Subscriptions: %s" , msg [ "stream" ] , list ( self . subscriptions . keys ( ) ) )
This function is called whenever there is a message received from the server
This function is called whenever there is a message received from the server