Watch the event system and return job data as it comes in

        :returns: all of the information for the JID
197634
def get_iter_returns(
            self,
            jid,
            minions,
            timeout=None,
            tgt='*',
            tgt_type='glob',
            expect_minions=False,
            block=True,
            **kwargs):
        '''
        
        '''
        if not isinstance(minions, set):
            if isinstance(minions, six.string_types):
                minions = set([minions])
            elif isinstance(minions, (list, tuple)):
                minions = set(list(minions))

        if timeout is None:
            timeout = self.opts['timeout']
        gather_job_timeout = int(kwargs.get('gather_job_timeout', self.opts['gather_job_timeout']))
        start = int(time.time())

        # timeouts per minion, id_ -> timeout time
        minion_timeouts = {}

        found = set()
        missing = set()
        # Check to see if the jid is real, if not return the empty dict
        try:
            if self.returners['{0}.get_load'.format(self.opts['master_job_cache'])](jid) == {}:
                log.warning('jid does not exist')
                yield {}
                # stop the iteration, since the jid is invalid
                raise StopIteration()
        except Exception as exc:
            log.warning('Returner unavailable: %s', exc, exc_info_on_loglevel=logging.DEBUG)
        # Wait for the hosts to check in
        last_time = False
        # iterator for this job's return
        if self.opts['order_masters']:
            # If we are a MoM, we need to gather expected minions from downstreams masters.
            ret_iter = self.get_returns_no_block('(salt/job|syndic/.*)/{0}'.format(jid), 'regex')
        else:
            ret_iter = self.get_returns_no_block('salt/job/{0}'.format(jid))
        # iterator for the info of this job
        jinfo_iter = []
        # open event jids that need to be un-subscribed from later
        open_jids = set()
        timeout_at = time.time() + timeout
        gather_syndic_wait = time.time() + self.opts['syndic_wait']
        # are there still minions running the job out there
        # start as True so that we ping at least once
        minions_running = True
        log.debug(
            'get_iter_returns for jid %s sent to %s will timeout at %s',
            jid, minions, datetime.fromtimestamp(timeout_at).time()
        )
        while True:
            # Process events until timeout is reached or all minions have returned
            for raw in ret_iter:
                # if we got None, then there were no events
                if raw is None:
                    break
                if 'minions' in raw.get('data', {}):
                    minions.update(raw['data']['minions'])
                    if 'missing' in raw.get('data', {}):
                        missing.update(raw['data']['missing'])
                    continue
                if 'return' not in raw['data']:
                    continue
                if kwargs.get('raw', False):
                    found.add(raw['data']['id'])
                    yield raw
                else:
                    found.add(raw['data']['id'])
                    ret = {raw['data']['id']: {'ret': raw['data']['return']}}
                    if 'out' in raw['data']:
                        ret[raw['data']['id']]['out'] = raw['data']['out']
                    if 'retcode' in raw['data']:
                        ret[raw['data']['id']]['retcode'] = raw['data']['retcode']
                    if 'jid' in raw['data']:
                        ret[raw['data']['id']]['jid'] = raw['data']['jid']
                    if kwargs.get('_cmd_meta', False):
                        ret[raw['data']['id']].update(raw['data'])
                    log.debug('jid %s return from %s', jid, raw['data']['id'])
                    yield ret

            # if we have all of the returns (and we aren't a syndic), no need for anything fancy
            if len(found.intersection(minions)) >= len(minions) and not self.opts['order_masters']:
                # All minions have returned, break out of the loop
                log.debug('jid %s found all minions %s', jid, found)
                break
            elif len(found.intersection(minions)) >= len(minions) and self.opts['order_masters']:
                if len(found) >= len(minions) and len(minions) > 0 and time.time() > gather_syndic_wait:
                    # There were some minions to find and we found them
                    # However, this does not imply that *all* masters have yet responded with expected minion lists.
                    # Therefore, continue to wait up to the syndic_wait period (calculated in gather_syndic_wait) to see
                    # if additional lower-level masters deliver their lists of expected
                    # minions.
                    break
            # If we get here we may not have gathered the minion list yet. Keep waiting
            # for all lower-level masters to respond with their minion lists

            # let start the timeouts for all remaining minions

            for id_ in minions - found:
                # if we have a new minion in the list, make sure it has a timeout
                if id_ not in minion_timeouts:
                    minion_timeouts[id_] = time.time() + timeout

            # if the jinfo has timed out and some minions are still running the job
            # re-do the ping
            if time.time() > timeout_at and minions_running:
                # since this is a new ping, no one has responded yet
                jinfo = self.gather_job_info(jid, list(minions - found), 'list', **kwargs)
                minions_running = False
                # if we weren't assigned any jid that means the master thinks
                # we have nothing to send
                if 'jid' not in jinfo:
                    jinfo_iter = []
                else:
                    jinfo_iter = self.get_returns_no_block('salt/job/{0}'.format(jinfo['jid']))
                timeout_at = time.time() + gather_job_timeout
                # if you are a syndic, wait a little longer
                if self.opts['order_masters']:
                    timeout_at += self.opts.get('syndic_wait', 1)

            # check for minions that are running the job still
            for raw in jinfo_iter:
                # if there are no more events, lets stop waiting for the jinfo
                if raw is None:
                    break
                try:
                    if raw['data']['retcode'] > 0:
                        log.error('saltutil returning errors on minion %s', raw['data']['id'])
                        minions.remove(raw['data']['id'])
                        break
                except KeyError as exc:
                    # This is a safe pass. We're just using the try/except to
                    # avoid having to deep-check for keys.
                    missing_key = exc.__str__().strip('\'"')
                    if missing_key == 'retcode':
                        log.debug('retcode missing from client return')
                    else:
                        log.debug(
                            'Passing on saltutil error. Key \'%s\' missing '
                            'from client return. This may be an error in '
                            'the client.', missing_key
                        )
                # Keep track of the jid events to unsubscribe from later
                open_jids.add(jinfo['jid'])

                # TODO: move to a library??
                if 'minions' in raw.get('data', {}):
                    minions.update(raw['data']['minions'])
                    continue
                if 'syndic' in raw.get('data', {}):
                    minions.update(raw['syndic'])
                    continue
                if 'return' not in raw.get('data', {}):
                    continue

                # if the job isn't running there anymore... don't count
                if raw['data']['return'] == {}:
                    continue

                # if the minion throws an exception containing the word "return"
                # the master will try to handle the string as a dict in the next
                # step. Check if we have a string, log the issue and continue.
                if isinstance(raw['data']['return'], six.string_types):
                    log.error("unexpected return from minion: %s", raw)
                    continue

                if 'return' in raw['data']['return'] and \
                    raw['data']['return']['return'] == {}:
                    continue

                # if we didn't originally target the minion, lets add it to the list
                if raw['data']['id'] not in minions:
                    minions.add(raw['data']['id'])
                # update this minion's timeout, as long as the job is still running
                minion_timeouts[raw['data']['id']] = time.time() + timeout
                # a minion returned, so we know its running somewhere
                minions_running = True

            # if we have hit gather_job_timeout (after firing the job) AND
            # if we have hit all minion timeouts, lets call it
            now = time.time()
            # if we have finished waiting, and no minions are running the job
            # then we need to see if each minion has timedout
            done = (now > timeout_at) and not minions_running
            if done:
                # if all minions have timeod out
                for id_ in minions - found:
                    if now < minion_timeouts[id_]:
                        done = False
                        break
            if done:
                break

            # don't spin
            if block:
                time.sleep(0.01)
            else:
                yield

        # If there are any remaining open events, clean them up.
        if open_jids:
            for jid in open_jids:
                self.event.unsubscribe(jid)

        if expect_minions:
            for minion in list((minions - found)):
                yield {minion: {'failed': True}}

        # Filter out any minions marked as missing for which we received
        # returns (prevents false events sent due to higher-level masters not
        # knowing about lower-level minions).
        missing -= found

        # Report on missing minions
        if missing:
            for minion in missing:
                yield {minion: {'failed': True}}
581553
def get(self, *jids):
        ''''''
        if jids:
            return [
                Job(self.client, **j) for j in
                json.loads(self.client('multiget', *jids))]
        return []
964754
async def _watch(self):
        

        file_name = os.path.basename(self._file_path)
        logger.info(
            'Watching %s "%s"',
            self.THING,
            self._file_path,
        )

        while self._running:
            evt = await self._watcher.get_event()

            if evt.name == file_name:
                await self._load()
                logger.info(
                    'Reloading changed %s from "%s"',
                    self.THING,
                    self._file_path
                )
466474
def check_job(cls, job_details):
        
        return check_log(job_details.logfile, cls.string_exited, cls.string_successful)
1180834
async def result_info(self) -> Optional[JobResult]:
        
        v = await self._redis.get(result_key_prefix + self.job_id, encoding=None)
        if v:
            return unpickle_result(v)
1005379
def get_cli_static_event_returns(
            self,
            jid,
            minions,
            timeout=None,
            tgt='*',
            tgt_type='glob',
            verbose=False,
            show_timeout=False,
            show_jid=False):
        '''
        
        '''
        log.trace('entered - function get_cli_static_event_returns()')
        minions = set(minions)
        if verbose:
            msg = 'Executing job with jid {0}'.format(jid)
            print(msg)
            print('-' * len(msg) + '\n')
        elif show_jid:
            print('jid: {0}'.format(jid))

        if timeout is None:
            timeout = self.opts['timeout']

        start = int(time.time())
        timeout_at = start + timeout
        found = set()
        ret = {}
        # Check to see if the jid is real, if not return the empty dict
        try:
            if self.returners['{0}.get_load'.format(self.opts['master_job_cache'])](jid) == {}:
                log.warning('jid does not exist')
                return ret
        except Exception as exc:
            raise SaltClientError('Load could not be retrieved from '
                                  'returner {0}. Exception details: {1}'.format(
                                      self.opts['master_job_cache'],
                                      exc))
        # Wait for the hosts to check in
        while True:
            # Process events until timeout is reached or all minions have returned
            time_left = timeout_at - int(time.time())
            # Wait 0 == forever, use a minimum of 1s
            wait = max(1, time_left)
            jid_tag = 'salt/job/{0}'.format(jid)
            raw = self.event.get_event(wait, jid_tag, auto_reconnect=self.auto_reconnect)
            if raw is not None and 'return' in raw:
                if 'minions' in raw.get('data', {}):
                    minions.update(raw['data']['minions'])
                    continue
                found.add(raw['id'])
                ret[raw['id']] = {'ret': raw['return']}
                ret[raw['id']]['success'] = raw.get('success', False)
                if 'out' in raw:
                    ret[raw['id']]['out'] = raw['out']
                if len(found.intersection(minions)) >= len(minions):
                    # All minions have returned, break out of the loop
                    break
                continue
            # Then event system timeout was reached and nothing was returned
            if len(found.intersection(minions)) >= len(minions):
                # All minions have returned, break out of the loop
                break
            if int(time.time()) > timeout_at:
                if verbose or show_timeout:
                    if self.opts.get('minion_data_cache', False) \
                            or tgt_type in ('glob', 'pcre', 'list'):
                        if len(found) < len(minions):
                            fail = sorted(list(minions.difference(found)))
                            for minion in fail:
                                ret[minion] = {
                                    'out': 'no_return',
                                    'ret': 'Minion did not return'
                                }
                break
            time.sleep(0.01)

        self._clean_up_subscriptions(jid)
        return ret
8275
def get_all_client_events(self, client):
        
        data = self._request('GET', '/events/{}'.format(client))
        return data.json()
339508
def _syndic_return(self, load):
        '''
        
        '''
        # Verify the load
        if any(key not in load for key in ('return', 'jid', 'id')):
            return None
        # if we have a load, save it
        if 'load' in load:
            fstr = '{0}.save_load'.format(self.opts['master_job_cache'])
            self.mminion.returners[fstr](load['jid'], load['load'])

        # Format individual return loads
        for key, item in six.iteritems(load['return']):
            ret = {'jid': load['jid'],
                   'id': key,
                   'return': item}
            if 'out' in load:
                ret['out'] = load['out']
            self._return(ret)
25201
def history(self):
        
        response = self.get("history")
        entries = _load_atom_entries(response)
        if entries is None: return []
        jobs = []
        for entry in entries:
            job = Job(self.service, entry.title)
            jobs.append(job)
        return jobs
1160169
def get_load(jid):
    '''
    
    '''
    with _get_serv(ret=None, commit=True) as cur:

        sql = '''SELECT load FROM jids WHERE jid = %s;'''
        cur.execute(sql, (jid,))
        data = cur.fetchone()
        if data:
            return data[0]
        return {}
662749
def get_event(id, opts = {})
      data, _status_code, _headers = get_event_with_http_info(id, opts)
      return data
    end
129944
def fetch_job_messages(offset, job_id, opts = {})
      data, _status_code, _headers = fetch_job_messages_with_http_info(offset, job_id, opts)
      return data
    end
215748
def _return(self, load):
        '''
        
        '''
        # Generate EndTime
        endtime = salt.utils.jid.jid_to_time(salt.utils.jid.gen_jid(self.opts))
        # If the return data is invalid, just ignore it
        if any(key not in load for key in ('return', 'jid', 'id')):
            return False

        if load['jid'] == 'req':
            # The minion is returning a standalone job, request a jobid
            prep_fstr = '{0}.prep_jid'.format(self.opts['master_job_cache'])
            load['jid'] = self.mminion.returners[prep_fstr](nocache=load.get('nocache', False))

            # save the load, since we don't have it
            saveload_fstr = '{0}.save_load'.format(self.opts['master_job_cache'])
            self.mminion.returners[saveload_fstr](load['jid'], load)
        log.info('Got return from %s for job %s', load['id'], load['jid'])
        self.event.fire_event(load, load['jid'])  # old dup event
        self.event.fire_event(load, salt.utils.event.tagify([load['jid'], 'ret', load['id']], 'job'))
        self.event.fire_ret_load(load)
        if not self.opts['job_cache'] or self.opts.get('ext_job_cache'):
            return

        fstr = '{0}.update_endtime'.format(self.opts['master_job_cache'])
        if (self.opts.get('job_cache_store_endtime')
                and fstr in self.mminion.returners):
            self.mminion.returners[fstr](load['jid'], endtime)

        fstr = '{0}.returner'.format(self.opts['master_job_cache'])
        self.mminion.returners[fstr](load)
1067097
def job_data(cls, job_id=dummy_job_id, status=None, code=None, error=None, **kwargs):
        
        return dict(job_id=job_id, status=status, code=code, error=error)
1297554
def events(self):
        
        if not self.__events:
            self.__events = Events(self.__connection)
        return self.__events
1098702
def event_callback(self, event_data):
        '''
        
        '''
        self.last_stdout_update = time.time()
        job_events_path = os.path.join(self.config.artifact_dir, 'job_events')
        if not os.path.exists(job_events_path):
            os.mkdir(job_events_path, 0o700)
        if 'uuid' in event_data:
            filename = '{}-partial.json'.format(event_data['uuid'])
            partial_filename = os.path.join(self.config.artifact_dir,
                                            'job_events',
                                            filename)
            full_filename = os.path.join(self.config.artifact_dir,
                                         'job_events',
                                         '{}-{}.json'.format(event_data['counter'],
                                                             event_data['uuid']))
            try:
                event_data.update(dict(runner_ident=str(self.config.ident)))
                try:
                    with codecs.open(partial_filename, 'r', encoding='utf-8') as read_file:
                        partial_event_data = json.load(read_file)
                    event_data.update(partial_event_data)
                    if self.remove_partials:
                        os.remove(partial_filename)
                except IOError:
                    debug("Failed to open ansible stdout callback plugin partial data file {}".format(partial_filename))
                if self.event_handler is not None:
                    should_write = self.event_handler(event_data)
                else:
                    should_write = True
                for plugin in ansible_runner.plugins:
                    ansible_runner.plugins[plugin].event_handler(self.config, event_data)
                if should_write:
                    with codecs.open(full_filename, 'w', encoding='utf-8') as write_file:
                        os.chmod(full_filename, stat.S_IRUSR | stat.S_IWUSR)
                        json.dump(event_data, write_file)
            except IOError as e:
                debug("Failed writing event data: {}".format(e))
1052438
def get_jid(jid):
    '''
    
    '''
    serv = _get_serv(ret=None)
    ret = {}
    for minion, data in six.iteritems(serv.hgetall('ret:{0}'.format(jid))):
        if data:
            ret[minion] = salt.utils.json.loads(data)
    return ret