docstr: When a message is received, send it to the `_handle` method, or
  await the incoming binary from the server.
docstr_tok: ["When","a","message","is","received",",","send","it","to","the","`","_handle","`","method",",","or","await","the","incoming","binary","from","the","server","."]
code: def on_message(self, msg):
      """When a message is received, send it to the `_handle` method, or
      await the incoming binary from the server."""
  
      if msg == "pong":
          # Do not respond to server pong heartbeats - only send them
          return
  
      self._lock.acquire()
      if self._pending_binary is not None and isinstance(msg, str):
          self._json_queue.append(msg)
  
      elif self._pending_binary is not None:
          binary_msg = msg
  
          self._full_binary += binary_msg
  
          if len(self._full_binary) == self._pending_binary_length:
              # Chunking is complete
              binary_msg = self._full_binary
  
          else:
              # Wait for the next chunk
              self._lock.release()
              return
  
          result = {
              "data": {"id": self._pending_binary, "data": binary_msg}
          }
  
          if self._pending_port_id is not None:
              result["data"]["data"] = {
                  "port_id": self._pending_port_id,
                  "delta": binary_msg,
              }
  
          # Clear flags for special binary message flow
          self._pending_binary = None
          self._pending_binary_length = None
          self._pending_port_id = None
          self._full_binary = b""
  
          # self._lock.release()
          self._handle(result)
          for msg in self._json_queue:
              msg = json.loads(msg)
              if msg.get("binary_length"):
                  self._pending_binary = msg["id"]
                  self._pending_binary_length = msg["binary_length"]
  
                  if (
                      msg.get("data")
                      and msg["data"].get("port_id", None) is not None
                  ):
                      self._pending_port_id = msg["data"].get("port_id")
              else:
                  self._handle({"data": msg})
  
          self._json_queue = []
  
      elif isinstance(msg, str):
          msg = json.loads(msg)
  
          if msg.get("binary_length"):
              self._pending_binary = msg["id"]
              self._pending_binary_length = msg["binary_length"]
  
              if (
                  msg.get("data")
                  and msg["data"].get("port_id", None) is not None
              ):
                  self._pending_port_id = msg["data"].get("port_id")
          else:
              self._handle({"data": msg})
      else:
          # websocket client sometimes calls None on disconnect ..
          pass
      self._lock.release()
  
file_path: perspective/python/perspective/perspective/client/websocket.py
imports: import asyncio
  import json
  from abc import ABC, abstractmethod
  from threading import Lock
  from .base import PerspectiveClient
  from ..manager.manager_internal import DateTimeEncoder
label_desc: Use image node_id 2 for calling the PerspectiveWebsocketClient obj's underlying member method code with example usage: obj.on_message(msg) without return types