:param plugin_name: <str>
    :param verify_job: <bool>
    :param conn: <connection> or <NoneType>
    :return: <generator> yields <dict>
197594
def get_jobs(plugin_name,
             verify_job=True, conn=None):
    
    job_cur = _jobs_cursor(plugin_name).run(conn)
    for job in job_cur:
        if verify_job and not verify(job, Job()):
            continue #to the next job... warn?
        yield job
709207
def present(name,
            config=None,
            **kwargs):
    '''
    
    '''

    ret = {'name': name,
           'result': True,
           'changes': {},
           'comment': ['Job {0} is up to date.'.format(name)]}

    if __salt__['jenkins.job_exists'](name):
        _current_job_config = __salt__['jenkins.get_job_config'](name)
        buf = six.moves.StringIO(_current_job_config)
        oldXML = ET.fromstring(buf.read())

        cached_source_path = __salt__['cp.cache_file'](config, __env__)
        with salt.utils.files.fopen(cached_source_path) as _fp:
            newXML = ET.fromstring(salt.utils.stringutils.to_unicode(_fp.read()))
        if not _elements_equal(oldXML, newXML):
            diff = difflib.unified_diff(
                ET.tostringlist(oldXML, encoding='utf8', method='xml'),
                ET.tostringlist(newXML, encoding='utf8', method='xml'), lineterm='')
            try:
                __salt__['jenkins.update_job'](name, config, __env__)
            except CommandExecutionError as exc:
                return _fail(ret, exc.strerror)
            else:
                ret['changes'] = ''.join(diff)
                ret['comment'].append('Job \'{0}\' updated.'.format(name))

    else:
        cached_source_path = __salt__['cp.cache_file'](config, __env__)
        with salt.utils.files.fopen(cached_source_path) as _fp:
            new_config_xml = salt.utils.stringutils.to_unicode(_fp.read())

        try:
            __salt__['jenkins.create_job'](name, config, __env__)
        except CommandExecutionError as exc:
            return _fail(ret, exc.strerror)

        buf = six.moves.StringIO(new_config_xml)
        diff = difflib.unified_diff('', buf.readlines(), lineterm='')
        ret['changes'][name] = ''.join(diff)
        ret['comment'].append('Job \'{0}\' added.'.format(name))

    ret['comment'] = '\n'.join(ret['comment'])
    return ret
205633
def _get_conn(ret):
    '''
    
    '''
    _options = _get_options(ret)

    host = _options.get('host')
    port = _options.get('port')
    db_ = _options.get('db')
    user = _options.get('user')
    password = _options.get('password')
    indexes = _options.get('indexes', False)

    # at some point we should remove support for
    # pymongo versions < 2.3 until then there are
    # a bunch of these sections that need to be supported

    if PYMONGO_VERSION > _LooseVersion('2.3'):
        conn = pymongo.MongoClient(host, port)
    else:
        conn = pymongo.Connection(host, port)
    mdb = conn[db_]

    if user and password:
        mdb.authenticate(user, password)

    if indexes:
        if PYMONGO_VERSION > _LooseVersion('2.3'):
            mdb.saltReturns.create_index('minion')
            mdb.saltReturns.create_index('jid')

            mdb.jobs.create_index('jid')
        else:
            mdb.saltReturns.ensure_index('minion')
            mdb.saltReturns.ensure_index('jid')

            mdb.jobs.ensure_index('jid')

    return conn, mdb
183797
def _parse_connection_name(self, name):
        
        if name is None:
            name = self.get_default_connection()

        if name.endswith(('::read', '::write')):
            return name.split('::', 1)

        return name, None
801624
def get_name(self, plugin):
        
        for name, val in self._name2plugin.items():
            if plugin == val:
                return name
318910
def job(self, name):
        
        for job in self.jobs():
            if job.data.name == name:
                return job
14675
def _jobs_cursor(plugin_name, location=None, port=None, custom=None):
    
    cur = RBJ.get_all(READY, index=STATUS_FIELD)
    cur_filter = (r.row[TARGET_FIELD][PLUGIN_NAME_KEY] == plugin_name)
    cur_filter = cur_filter & \
                 (~r.row.has_fields(EXPIRE_FIELD) |
                  r.row[EXPIRE_FIELD].ge(time()))
    if location:
        cur_filter = cur_filter & \
                     (r.row[TARGET_FIELD][LOCATION_FIELD] == location)
    if port:
        cur_filter = cur_filter & \
                     (r.row[TARGET_FIELD][PORT_FIELD] == port)
    if custom:
        cur_filter = cur_filter & custom
    return cur.filter(cur_filter).order_by(START_FIELD)
1030160
def get_targets_by_plugin(plugin_name, conn=None):
    
    targets = RBT
    results = targets.filter({PLUGIN_NAME_KEY: plugin_name}).run(conn)
    for item in results:
        yield item
988667
function ( name ) {
		// valid connection
		if ( name in pool ) {
			return {
				active: pool[name].active,
				count:  pool[name].count
			};
		}

		// failure
		return {active: false};
	}
34855
def onRefreshPluginData(self, plugin_name, data):
        
        logger.info(u"onRefreshPluginData: {}".format(plugin_name))
        if not plugin_name:
            logger.error("Missing plugin name")
            return
        reactor.callFromThread(self._sendJSON, {
            'msg': "plugin_data_get",
            'plugin_name': plugin_name
        })
917156
def from_json(cls, plugin):
        

        process = None
        repair = None

        name = plugin["name"] + "Proxy"
        cls = type(name, (cls,), plugin)

        # Emulate function
        for name in ("process", "repair"):
            args = ", ".join(plugin["process"]["args"])
            func = "def {name}({args}): pass".format(name=name,
                                                     args=args)
            exec(func)

        cls.process = process
        cls.repair = repair

        cls.__orig__ = plugin

        return cls
986095
def _process_plugin(self, plugin):
        '''
        
        '''
        instance = plugin['instance']
        regex = plugin['regex']
        for key in self.redis_conn.scan_iter(match=regex):
            # acquire lock
            lock = self._create_lock_object(key)

            try:
                if lock.acquire(blocking=False):
                    val = self.redis_conn.get(key)
                    self._process_key_val(instance, key, val)
            except Exception:
                self.logger.error(traceback.format_exc())
                self._increment_fail_stat('{k}:{v}'.format(k=key, v=val))

                self._process_failures(key)

            # remove lock regardless of if exception or was handled ok
            if lock._held:
                self.logger.debug("releasing lock")
                lock.release()
790168
def _plugin_get(self, plugin_name):
        
        if not plugin_name:
            return None, u"Plugin name not set"
        for plugin in self.controller.plugins:
            if not isinstance(plugin, SettablePlugin):
                continue
            if plugin.name == plugin_name:
                return plugin, ""
        return None, u"Settable plugin '{}' not found".format(plugin_name)
15929
def insert_jobs(jobs, verify_jobs=True, conn=None):
    
    assert isinstance(jobs, list)
    if verify_jobs \
            and not verify({Jobs.DESCRIPTOR.name: jobs}, Jobs()):
        raise ValueError("Invalid Jobs")
    inserted = RBJ.insert(jobs).run(conn)
    return inserted
650933
def push_plugin(self, name):
        
        url = self._url('/plugins/{0}/pull', name)

        headers = {}
        registry, repo_name = auth.resolve_repository_name(name)
        header = auth.get_config_header(self, registry)
        if header:
            headers['X-Registry-Auth'] = header
        res = self._post(url, headers=headers)
        self._raise_for_status(res)
        return self._stream_helper(res, decode=True)
245697
def connect_job(job_id,
                deployment_name,
                token_manager=None,
                app_url=defaults.APP_URL,
                persist=False,
                websocket=None,
                data_url=None):
    

    if data_url == None:
        data_url = get_data_url_for_job(job_id,
                                        deployment_name,
                                        token_manager=token_manager,
                                        app_url=app_url)

    if websocket == None:
        websocket = __wss_connect(data_url,
                                  token_manager,
                                  job_id=job_id)

    pong = json.dumps({
        'pong': True
    })

    if not persist:
        job_finished = False

        while not job_finished:
            try:
                data = websocket.recv()

                if data:
                    payload = json.loads(data)

                    if is_debug_enabled():
                        printable_payload = dict(payload)
                        if 'points' in payload:
                            # don't want to print out all the outputs when in
                            # debug mode
                            del printable_payload['points']
                            printable_payload['points'] = 'NOT SHOWN'

                        debug('received %s' % json.dumps(printable_payload))

                    if 'ping' in payload.keys():
                        # ping/pong (ie heartbeat) mechanism
                        websocket.send(pong)

                        if is_debug_enabled():
                            debug('sent %s' % json.dumps(pong))

                    if 'job_end' in payload.keys() and payload['job_end'] == True:
                        job_finished = True

                    if token_manager.is_access_token_expired():
                        debug('refreshing access token')
                        token_obj = {
                            "accessToken": token_manager.get_access_token()
                        }
                        # refresh authentication token
                        websocket.send(json.dumps(token_obj))

                    if 'error' in payload:
                        if payload['error'] == 'NONEXISTENT-JOB':
                            raise JutException('Job "%s" no longer running' % job_id)

                    # return all channel messages
                    yield payload

                else:
                    debug('payload was "%s", forcing websocket reconnect' % data)
                    raise IOError()

            except IOError:
                if is_debug_enabled():
                    traceback.print_exc()
                #
                # We'll retry for just under 30s since internally we stop
                # running non persistent programs after 30s of not heartbeating
                # with the client
                #
                retry = 1
                while retry <= 5:
                    try:
                        debug('network error reconnecting to job %s, '
                              'try %s of 5' % (job_id, retry))
                        websocket = __wss_connect(data_url, token_manager, job_id=job_id)
                        break

                    except socket.error:

                        if is_debug_enabled():
                            traceback.print_exc()

                        retry += 1
                        time.sleep(5)

                debug('network error reconnecting to job %s, '
                      'try %s of 5' % (job_id, retry))
                websocket = __wss_connect(data_url, token_manager, job_id=job_id)

    websocket.close()
248792
def show_network(kwargs=None, call=None):
    '''
    
    '''
    if call != 'function':
        raise SaltCloudSystemExit(
            'The show_network function must be called with -f or --function.'
        )
    if not kwargs or 'name' not in kwargs:
        log.error(
            'Must specify name of network.'
        )
        return False

    conn = get_conn()
    return _expand_item(conn.ex_get_network(kwargs['name']))