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):
        """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()))
This function is called whenever there is a message received from the server