code: def _receive_obj(self, channel_info, name, tag):
      party_id = channel_info._dst_party_id
      role = channel_info._dst_role
  
      wish_cache_key = _get_message_cache_key(name, tag, party_id, role)
  
      if wish_cache_key in self._message_cache:
          recv_bytes = self._message_cache[wish_cache_key]
          del self._message_cache[wish_cache_key]
          return recv_bytes
  
      # channel_info = self._query_receive_topic(channel_info)
  
      for _id, properties, body in self._get_consume_message(
          channel_info
      ):
          LOGGER.debug(f"properties: {properties}")
          cache_key = _get_message_cache_key(
              properties["message_id"],
              properties["correlation_id"],
              party_id,
              role,
          )
          # object
          if properties["content_type"] == "text/plain":
              recv_bytes = body
              self._consume_ack(channel_info, _id)
              LOGGER.debug(
                  f"[federation._receive_obj] cache_key: {cache_key}, wish_cache_key: {wish_cache_key}"
              )
              if cache_key == wish_cache_key:
                  channel_info.cancel()
                  return recv_bytes
              else:
                  self._message_cache[cache_key] = recv_bytes
          else:
              raise ValueError(
                  f"[federation._receive_obj] properties.content_type is {properties['content_type']}, but must be text/plain"
              )
  
imports: import json
  import logging
  import sys
  import typing
  from typing import List
  from fate.arch.computing.api import KVTableContext
  from fate.arch.federation.api import Federation, PartyMeta, TableMeta
  from ._datastream import Datastream
  from ._parties import Party
label_desc: Use image node_id 23 for calling the MessageQueueBasedFederation obj's underlying member method code with example usage: obj._receive_obj(channel_info, name, tag) and returns: recv_bytes, recv_bytes