@tornado.gen.coroutine
    def pub_async(self,
                  tgt,
                  fun,
                  arg=(),
                  expr_form='glob',
                  ret='',
                  jid='',
                  timeout=5,
                  io_loop=None,
                  listen=True,
                  **kwargs):
        # Make sure the publisher is running by checking the unix socket
        if (self.opts.get('ipc_mode', '') != 'tcp' and
                not os.path.exists(os.path.join(self.opts['sock_dir'],
                'publish_pull.ipc'))):
            log.error(
                'Unable to connect to the salt master publisher at '
                '{0}'.format(self.opts['sock_dir'])
            )
            raise SaltClientError

        payload_kwargs = self._prep_pub(
                tgt,
                fun,
                arg,
                expr_form,
                ret,
                jid,
                timeout,
                **kwargs)

        master_uri = 'tcp://' + salt.utils.ip_bracket(self.opts['interface']) + \
                     ':' + str(self.opts['ret_port'])
        channel = salt.transport.client.AsyncReqChannel.factory(self.opts,
                                                                io_loop=io_loop,
                                                                crypt='clear',
                                                                master_uri=master_uri)

        try:
            # Ensure that the event subscriber is connected.
            # If not, we won't get a response, so error out
            if listen and not self.event.connect_pub(timeout=timeout):
                raise SaltReqTimeoutError()
            payload = yield channel.send(payload_kwargs, timeout=timeout)
        except SaltReqTimeoutError:
            raise SaltReqTimeoutError(
                'Salt request timed out. The master is not responding. '
                'If this error persists after verifying the master is up, '
                'worker_threads may need to be increased.'
            )

        if not payload:
            # The master key could have changed out from under us! Regen
            # and try again if the key has changed
            key = self.__read_master_key()
            if key == self.key:
                raise tornado.gen.Return(payload)
            self.key = key
            payload_kwargs['key'] = self.key
            payload = yield channel.send(payload_kwargs)

        error = payload.pop('error', None)
        if error is not None:
            raise PublishError(error)

        if not payload:
            raise tornado.gen.Return(payload)

        # We have the payload, let's get rid of the channel fast(GC'ed faster)
        del channel

        raise tornado.gen.Return({'jid': payload['load']['jid'],
                                  'minions': payload['load']['minions']})
Take the required arguments and publish the given command.
dataset/ETHPy150Open saltstack/salt/salt/client/__init__.py/LocalClient.pub_async