def run_commands_from_slack_async(self, message_generator, fire_all, tag, control, interval=1):
        '''
        Pull any pending messages from the message_generator, sending each
        one to either the event bus, the command_async or both, depending on
        the values of fire_all and command
        '''

        outstanding = {}  # set of job_id that we need to check for

        while True:
            log.trace('Sleeping for interval of %s', interval)
            time.sleep(interval)
            # Drain the slack messages, up to 10 messages at a clip
            count = 0
            for msg in message_generator:
                # The message_generator yields dicts.  Leave this loop
                # on a dict that looks like {'done': True} or when we've done it
                # 10 times without taking a break.
                log.trace('Got a message from the generator: %s', msg.keys())
                if count > 10:
                    log.warning('Breaking in getting messages because count is exceeded')
                    break
                if not msg:
                    count += 1
                    log.warning('Skipping an empty message.')
                    continue  # This one is a dud, get the next message
                if msg.get('done'):
                    log.trace('msg is done')
                    break
                if fire_all:
                    log.debug('Firing message to the bus with tag: %s', tag)
                    log.debug('%s %s', tag, msg)
                    self.fire('{0}/{1}'.format(tag, msg['message_data'].get('type')), msg)
                if control and (len(msg) > 1) and msg.get('cmdline'):
                    channel = self.sc.server.channels.find(msg['channel'])
                    jid = self.run_command_async(msg)
                    log.debug('Submitted a job and got jid: %s', jid)
                    outstanding[jid] = msg  # record so we can return messages to the caller
                    channel.send_message("@{}'s job is submitted as salt jid {}".format(msg['user_name'], jid))
                count += 1
            start_time = time.time()
            job_status = self.get_jobs_from_runner(outstanding.keys())  # dict of job_ids:results are returned
            log.trace('Getting %s jobs status took %s seconds', len(job_status), time.time() - start_time)
            for jid in job_status:
                result = job_status[jid]['data']
                function = job_status[jid]['function']
                if result:
                    log.debug('ret to send back is %s', result)
                    # formatting function?
                    this_job = outstanding[jid]
                    channel = self.sc.server.channels.find(this_job['channel'])
                    return_text = self.format_return_text(result, function)
                    return_prefix = "@{}'s job `{}` (id: {}) (target: {}) returned".format(
                        this_job['user_name'], this_job['cmdline'], jid, this_job['target'])
                    channel.send_message(return_prefix)
                    ts = time.time()
                    st = datetime.datetime.fromtimestamp(ts).strftime('%Y%m%d%H%M%S%f')
                    filename = 'salt-results-{0}.yaml'.format(st)
                    r = self.sc.api_call(
                        'files.upload', channels=channel.id, filename=filename,
                        content=return_text)
                    # Handle unicode return
                    log.debug('Got back %s via the slack client', r)
                    resp = salt.utils.yaml.safe_load(salt.utils.json.dumps(r))
                    if 'ok' in resp and resp['ok'] is False:
                        this_job['channel'].send_message('Error: {0}'.format(resp['error']))
                    del outstanding[jid]