code: async def _call_async(self, method_name: str, *args, **kwargs):
          
          request = utils.rpc_request(method_name, *args, **kwargs)
          _log.debug("Sending request: %s", request)
  
          # setup an event to notify us when the reply is received (potentially by a task scheduled by
          # another call to _async_call). we do this before we send the request to catch the case
          # where the reply comes back before we re-enter this thread
          self._events[request.id] = asyncio.Event()
  
          # schedule a task to receive the reply to ensure we have a task to receive the reply
          asyncio.ensure_future(self._recv_reply())
  
          await self._async_socket.send_multipart([to_msgpack(request)])
          await self._events[request.id].wait()
  
          reply = self._replies.pop(request.id)
          if isinstance(reply, RPCError):
              raise utils.RPCError(reply.error)
          else:
              return reply.result
docs: Sends a request to the socket and then wait for the reply.
  
          To deal with multiple, asynchronous requests we do not expect that the receive reply task
          scheduled from this call is the one that receives this call's reply and instead rely on
          Events to signal across multiple _async_call/_recv_reply tasks.