def cache_resources(self, resource_type, specific_ids=None):
        """
        Save all resources of a given type to memory, for in-memory lookup ops

        This is a PREFETCH operation, and EnhancedResourceRegistryClient objects that use the cache functionality
        should NOT be kept across service calls.
        """
        #log.info("Caching resources: %s", resource_type)
        #log.debug("This cache is %s", self)
        time_caching_start = get_ion_ts()

        resource_objs = []
        if specific_ids is None:
            resource_objs, _ = self.RR.find_resources(restype=resource_type, id_only=False)
        else:
            assert type(specific_ids) is list
            if specific_ids:
                resource_objs = self.RR.read_mult(specific_ids)

        lookups = DotDict()
        lookups.by_id =   {}
        lookups.by_name = {}
        self._cached_resources[resource_type] = lookups

        for r in resource_objs:
            self._add_resource_to_cache(resource_type, r)

        time_caching_stop = get_ion_ts()

        total_time = int(time_caching_stop) - int(time_caching_start)

    def take_snapshot(self, snapshot_id=None, include_list=None, exclude_list=None, snapshot_kwargs=None):
        if include_list:
            self.snapshots.add(include_list)
        if exclude_list:
            for item in exclude_list:
                self.snapshots.remove(item)
        if not snapshot_id:
            snapshot_id = get_ion_ts()
        if not snapshot_kwargs:
            snapshot_kwargs = {}

        self.snapshot["snapshot_ts_begin"] = get_ion_ts()
        self.snapshot["snapshot_list"] = self.snapshots
        for snap in self.snapshots:
            snap_func = "_snap_%s" % snap
            func = getattr(self, snap_func, None)
            if func:
                try:
                    snap_result = func(**snapshot_kwargs)
                except Exception as ex:
                    log.warn("Could not take snapshot %s: %s" % (snap, str(ex)))
                self.snapshot[snap] = snap_result
            else:
                log.warn("Snapshot function %s undefined" % snap_func)

        self.snap_ts = get_ion_ts()
        self.snapshot["snapshot_ts"] = self.snap_ts
        self.snapshot["snapshot_id"] = snapshot_id

    def test_get_valid_resource_commitment(self):
        from pyon.util.containers import get_ion_ts

        # create ION org and an actor
        ion_org = IonObject(RT.Org, name='ION')
        ion_org_id, _ = self.rr.create(ion_org)
        ion_org._id = ion_org_id
        actor = IonObject(RT.ActorIdentity, name='actor1')
        actor_id, _ = self.rr.create(actor)

        # create an expired commitment in the org
        ts = int(get_ion_ts()) - 50000
        com_obj = IonObject(RT.Commitment, provider=ion_org_id, consumer=actor_id, commitment=True, expiration=ts)
        com_id, _ = self.rr.create(com_obj)
        id = self.rr.create_association(ion_org_id, PRED.hasCommitment, com_id)
        c = get_valid_resource_commitments(ion_org_id, actor_id)
        #verify that the commitment is not returned
        self.assertIsNone(c)

        # create a commitment that has not expired yet
        ts = int(get_ion_ts()) + 50000
        com_obj = IonObject(RT.Commitment, provider=ion_org_id, consumer=actor_id, commitment=True, expiration=ts)
        com_id, _ = self.rr.create(com_obj)
        id = self.rr.create_association(ion_org_id, PRED.hasCommitment, com_id)
        c = get_valid_resource_commitments(ion_org_id, actor_id)

        #verify that the commitment is returned
        self.assertIsNotNone(c)

 def _create_dir_entry(self, object_id,  parent, key, attributes=None, ts_created='', ts_updated=''):
     doc = {}
     doc['_id'] = object_id
     doc['type_'] = 'DirEntry'
     doc['attributes'] = attributes or {}
     doc['key'] = key
     doc['parent'] = parent
     doc['ts_created'] = ts_created or get_ion_ts()
     doc['ts_updated'] = ts_updated or get_ion_ts()
     return doc

 def _create_dir_entry(self, object_id, parent, key, attributes=None, ts_created="", ts_updated=""):
     doc = {}
     doc["_id"] = object_id
     doc["type_"] = "DirEntry"
     doc["attributes"] = attributes or {}
     doc["key"] = key
     doc["parent"] = parent if parent else "/"
     doc["org"] = self.orgname
     doc["ts_created"] = ts_created or get_ion_ts()
     doc["ts_updated"] = ts_updated or get_ion_ts()
     return doc

 def put_state(self, key, state):
     log.debug("Store persistent state for key=%s" % key)
     if not isinstance(state, dict):
         raise BadRequest("state must be type dict, not %s" % type(state))
     try:
         state_obj = self.state_store.read(key)
         state_obj.state = state
         state_obj.ts = get_ion_ts()
         self.state_store.update(state_obj)
     except NotFound as nf:
         state_obj = ProcessState(state=state, ts=get_ion_ts())
         self.state_store.create(state_obj, object_id=key)

    def heartbeat(self):
        """
        Returns a tuple indicating everything is ok.

        Should only be called after the process has been started.
        Checks the following:
            - All attached endpoints are alive + listening (this means ready)
            - The control flow greenlet is alive + listening or processing

        @return 3-tuple indicating (listeners ok, ctrl thread ok, heartbeat status). Use all on it for a
                boolean indication of success.
        """
        listeners_ok = True
        for l in self.listeners:
            if not (l in self._listener_map and not self._listener_map[l].proc.dead and l.get_ready_event().is_set()):
                listeners_ok = False

        ctrl_thread_ok = self._ctrl_thread.running

        # are we currently processing something?
        heartbeat_ok = True
        if self._ctrl_current is not None:
            st = traceback.extract_stack(self._ctrl_thread.proc.gr_frame)

            if self._ctrl_current == self._heartbeat_op:

                if st == self._heartbeat_stack:
                    self._heartbeat_count += 1  # we've seen this before! increment count

                    # we've been in this for the last X ticks, or it's been X seconds, fail this part of the heartbeat
                    if (
                        self._heartbeat_count > CFG.get_safe("cc.timeout.heartbeat_proc_count_threshold", 30)
                        or int(get_ion_ts()) - int(self._heartbeat_time)
                        >= CFG.get_safe("cc.timeout.heartbeat_proc_time_threshold", 30) * 1000
                    ):
                        heartbeat_ok = False
                else:
                    # it's made some progress
                    self._heartbeat_count = 1
                    self._heartbeat_stack = st
                    self._heartbeat_time = get_ion_ts()
            else:
                self._heartbeat_op = self._ctrl_current
                self._heartbeat_count = 1
                self._heartbeat_time = get_ion_ts()
                self._heartbeat_stack = st

        else:
            self._heartbeat_op = None
            self._heartbeat_count = 0

        return (listeners_ok, ctrl_thread_ok, heartbeat_ok)

    def _update_cached_resources(self):
        # cache some resources for in-memory lookups
        rsrcs = self._resources_to_cache()
        log.debug("updating cached resources: %s" % rsrcs)
        time_caching_start = get_ion_ts()
        for r in rsrcs:
            log.debug(" - %s", r)
            self.RR2.cache_resources(r)
        time_caching_stop = get_ion_ts()

        total_time = int(time_caching_stop) - int(time_caching_start)

        log.info("Cached %s resource types in %s seconds", len(rsrcs), total_time / 1000.0)

    def _update_cached_predicates(self):
        # cache some predicates for in-memory lookups
        preds = self._predicates_to_cache()
        log.debug("updating cached predicates: %s" % preds)
        time_caching_start = get_ion_ts()
        for pred in preds:
            log.debug(" - %s", pred)
            self.RR2.cache_predicate(pred)
        time_caching_stop = get_ion_ts()

        total_time = int(time_caching_stop) - int(time_caching_start)

        log.info("Cached %s predicates in %s seconds", len(preds), total_time / 1000.0)

    def test_pub_and_sub(self):
        ar = event.AsyncResult()
        gq = queue.Queue()
        self.count = 0

        def cb(*args, **kwargs):
            self.count += 1
            gq.put(args[0])
            if self.count == 2:
                ar.set()

        sub = EventSubscriber(event_type="ResourceEvent", callback=cb, origin="specific")
        pub = EventPublisher(event_type="ResourceEvent")

        self._listen(sub)
        pub.publish_event(origin="specific", description="hello")

        event_obj = bootstrap.IonObject('ResourceEvent', origin='specific', description='more testing')
        self.assertEqual(event_obj, pub.publish_event_object(event_obj))

        with self.assertRaises(BadRequest) as cm:
            event_obj = bootstrap.IonObject('ResourceEvent', origin='specific', description='more testing', ts_created='2423')
            pub.publish_event_object(event_obj)
        self.assertIn( 'The ts_created value is not a valid timestamp',cm.exception.message)

        with self.assertRaises(BadRequest) as cm:
            event_obj = bootstrap.IonObject('ResourceEvent', origin='specific', description='more testing', ts_created='1000494978462')
            pub.publish_event_object(event_obj)
        self.assertIn( 'This ts_created value is too old',cm.exception.message)

        with self.assertRaises(BadRequest) as cm:
            event_obj = bootstrap.IonObject('ResourceEvent', origin='specific', description='more testing')
            event_obj._id = '343434'
            pub.publish_event_object(event_obj)
        self.assertIn( 'The event object cannot contain a _id field',cm.exception.message)

        ar.get(timeout=5)

        res = []
        for x in xrange(self.count):
            res.append(gq.get(timeout=5))

        self.assertEquals(len(res), self.count)
        self.assertEquals(res[0].description, "hello")
        self.assertAlmostEquals(int(res[0].ts_created), int(get_ion_ts()), delta=5000)

        self.assertEquals(res[1].description, "more testing")
        self.assertAlmostEquals(int(res[1].ts_created), int(get_ion_ts()), delta=5000)

    def retire(self, resource_id):
        """
        This is the official "delete" for resource objects: they are set to RETIRED lcstate.
        All associations are set to retired as well.
        """
        res_obj = self.read(resource_id)
        old_state = res_obj.lcstate
        old_availability = res_obj.availability
        if old_state == LCS.RETIRED:
            raise BadRequest("Resource id=%s already RETIRED" % (resource_id))

        res_obj.lcstate = LCS.RETIRED
        res_obj.availability = AS.PRIVATE
        res_obj.ts_updated = get_ion_ts()

        updres = self.rr_store.update(res_obj)
        log.debug("retire(res_id=%s). Change %s_%s to %s_%s", resource_id,
                  old_state, old_availability, res_obj.lcstate, res_obj.availability)

        assocs = self.find_associations(anyside=resource_id, id_only=False)
        for assoc in assocs:
            assoc.retired = True
        if assocs:
            self.rr_store.update_mult(assocs)
            log.debug("retire(res_id=%s). Retired %s associations", resource_id, len(assocs))

        if self.container.has_capability(self.container.CCAP.EVENT_PUBLISHER):
            self.event_pub.publish_event(event_type="ResourceLifecycleEvent",
                                     origin=res_obj._id, origin_type=res_obj.type_,
                                     sub_type="%s.%s" % (res_obj.lcstate, res_obj.availability),
                                     lcstate=res_obj.lcstate, availability=res_obj.availability,
                                     lcstate_before=old_state, availability_before=old_availability)

    def create(self, object=None, actor_id=None):
        if object is None:
            raise BadRequest("Object not present")
        if not isinstance(object, IonObjectBase):
            raise BadRequest("Object is not an IonObject")
        if not is_resource(object):
            raise BadRequest("Object is not a Resource")

        lcsm = get_restype_lcsm(object._get_type())
        object.lcstate = lcsm.initial_state if lcsm else "DEPLOYED_AVAILABLE"
        cur_time = get_ion_ts()
        object.ts_created = cur_time
        object.ts_updated = cur_time
        new_res_id = create_unique_resource_id()
        res = self.rr_store.create(object, new_res_id)
        res_id, rev = res

        if actor_id and actor_id != 'anonymous':
            log.debug("Associate resource_id=%s with owner=%s" % (res_id, actor_id))
            self.rr_store.create_association(res_id, PRED.hasOwner, actor_id)

        self.event_pub.publish_event(event_type="ResourceModifiedEvent",
                                     origin=res_id, origin_type=object._get_type(),
                                     sub_type="CREATE",
                                     mod_type=ResourceModificationType.CREATE)

        return res

    def _execute(self, cprefix, command):
        if not command:
            raise iex.BadRequest("execute argument 'command' not present")
        if not command.command:
            raise iex.BadRequest("command not set")

        cmd_res = IonObject("AgentCommandResult", command_id=command.command_id, command=command.command)
        cmd_func = getattr(self, cprefix + str(command.command), None)
        if cmd_func:
            cmd_res.ts_execute = get_ion_ts()
            try:
                res = cmd_func(*command.args, **command.kwargs)
                cmd_res.status = 0
                cmd_res.result = res
            except Exception as ex:
                # TODO: Distinguish application vs. uncaught exception
                cmd_res.status = getattr(ex, 'status_code', -1)
                cmd_res.result = str(ex)
                log.info("Agent function failed with ex=%s msg=%s" % (type(ex), str(ex)))
        else:
            log.info("Agent command not supported: %s" % (command.command))
            ex = iex.NotFound("Command not supported: %s" % command.command)
            cmd_res.status = iex.NotFound.status_code
            cmd_res.result = str(ex)
        return cmd_res

    def is_resource_acquired_exclusively(self, actor_id='', resource_id=''):
        """Returns True if the specified resource_id has been acquired exclusively. The actor_id is optional, as the operation can
        return True if the resource is acquired exclusively by any actor or specifically by the specified actor_id,
        otherwise False is returned.

        @param actor_id    str
        @param resource_id    str
        @retval success    bool
        @throws BadRequest    if resource_id is not specified
        """
        if not resource_id:
            raise BadRequest("The resource_id parameter is missing")

        try:
            cur_time = int(get_ion_ts())
            commitments,_ = self.clients.resource_registry.find_objects(resource_id,PRED.hasCommitment, RT.Commitment)
            if commitments:
                for com in commitments:
                    if com.lcstate == LCS.RETIRED: #TODO remove when RR find_objects does not include retired objects
                        continue

                    #If the expiration is not 0 make sure it has not expired
                    if ( actor_id is None or actor_id == com.consumer )  and com.commitment.exclusive and\
                       int(com.expiration) > 0 and cur_time < int(com.expiration):
                        return True

        except Exception, e:
            log.error('is_resource_acquired_exclusively: %s for actor_id:%s and resource_id:%s' %  (e.message, actor_id, resource_id))

    def _execute(self, cprefix, command):
        if not command:
            raise iex.BadRequest("execute argument 'command' not present")
        if not command.command:
            raise iex.BadRequest("command not set")

        cmd_res = IonObject("AgentCommandResult", command_id=command.command_id, command=command.command)
        cmd_func = getattr(self, cprefix + str(command.command), None)
        if cmd_func:
            cmd_res.ts_execute = get_ion_ts()
            try:
                res = cmd_func(*command.args, **command.kwargs)
                cmd_res.status = 0
                cmd_res.result = res
            except iex.IonException as ex:
                # TODO: Distinguish application vs. uncaught exception
                cmd_res.status = getattr(ex, 'status_code', -1)
                cmd_res.result = str(ex)
                log.warn("Agent command %s failed with trace=%s" % (command.command, traceback.format_exc()))
        else:
            log.info("Agent command not supported: %s" % (command.command))
            ex = iex.NotFound("Command not supported: %s" % command.command)
            cmd_res.status = iex.NotFound.status_code
            cmd_res.result = str(ex)

        sub_type = "%s.%s" % (command.command, cmd_res.status)
        post_event = self._event_publisher._create_event(event_type=self.COMMAND_EVENT_TYPE,
                                origin=self.resource_id, origin_type=self.ORIGIN_TYPE,
                                sub_type=sub_type, command=command, result=cmd_res)
        post_event = self._post_execute_event_hook(post_event)
        success = self._event_publisher._publish_event(post_event, origin=post_event.origin)

        return cmd_res

    def set_lifecycle_state(self, resource_id='', target_lcstate=''):
        if not target_lcstate or target_lcstate not in LCS:
            raise BadRequest("Unknown life-cycle state %s" % target_lcstate)

        res_obj = self.read(resource_id)
        old_state = res_obj.lcstate
        if target_lcstate != LCS.RETIRED:
            restype = res_obj._get_type()
            restype_workflow = get_restype_lcsm(restype)
            if not restype_workflow:
                raise BadRequest("Resource id=%s type=%s has no lifecycle" % (resource_id, restype))

            # Check that target state is allowed
            if not target_lcstate in restype_workflow.get_successors(res_obj.lcstate).values():
                raise BadRequest("Target state %s not reachable for resource in state %s" % (target_lcstate, res_obj.lcstate))

        res_obj.lcstate = target_lcstate
        res_obj.ts_updated = get_ion_ts()

        updres = self.rr_store.update(res_obj)

        self.event_pub.publish_event(event_type="ResourceLifecycleEvent",
                                     origin=res_obj._id, origin_type=res_obj._get_type(),
                                     sub_type=target_lcstate,
                                     old_state=old_state, new_state=target_lcstate)

    def register(self, parent, key, **kwargs):
        """
        Add/replace an entry to directory below a parent node.
        Note: Does not merge the attribute values of the entry if existing
        """
        if not (parent and key):
            raise BadRequest("Illegal arguments")
        if not type(parent) is str or not parent.startswith("/"):
            raise BadRequest("Illegal arguments: parent")

        dn = self._get_dn(parent, key)
        log.debug("Directory.add(%s): %s" % (dn, kwargs))

        entry_old = None
        direntry = self._safe_read(dn)
        cur_time = get_ion_ts()
        if direntry:
            entry_old = direntry.attributes
            direntry.attributes = kwargs
            direntry.ts_updated=cur_time
            # TODO: This may fail because of concurrent update
            self.dir_store.update(direntry)
        else:
            parent_dn = self._get_dn(parent)
            direntry = DirEntry(parent=parent_dn, key=key, attributes=kwargs, ts_created=cur_time, ts_updated=cur_time)
            self.dir_store.create(direntry, dn)

        return entry_old

 def sample_other_op(self, foo='bar', num=84, name=''):
     log.info("StatefulTestProcess OP, state=%s", foo)
     newstate = foo
     oldstate = self._get_state("statevalue") or ""
     self._set_state("statevalue", newstate)
     self._set_state("statets", get_ion_ts())
     return oldstate

    def _on_command_error(self, cmd, execute_cmd, args, kwargs, ex):
        log.info('Resource agent %s publishing command error event: \
                 cmd=%s, execute_cmd=%s, args=%s, kwargs=%s, \
                 errtype=%s, errmsg=%s, errno=%i, time=%s',
                 self.id, str(cmd), str(execute_cmd), str(args), str(kwargs),
                 str(type(ex)), ex.message, ex.status_code, get_ion_ts())

        if hasattr(ex, 'status_code'):
            status_code = ex.status_code
        else:
            status_code = -1

        event_data = {
            'command': cmd,
            'execute_command': execute_cmd,
            'args': args,
            'kwargs': kwargs,
            'error_type': str(type(ex)),
            'error_msg': ex.message,
            'error_code': status_code
        }

        self._event_publisher.publish_event(event_type='ResourceAgentErrorEvent',
                                            origin=self.resource_id,
                                            **event_data)

    def _finalize_uirefs(self, ds):
        # Create real resource IDs
        for obj in self.ui_objs.values():
            oid = self.uiid_prefix + obj.uirefid
            obj._id = oid
            self.ui_obj_by_id[oid] = obj

            # Change references for all known UI objects
            for attr in obj.__dict__:
                if attr != 'uirefid' and getattr(obj, attr) in self.ui_objs:
                    setattr(obj, attr, self.uiid_prefix + getattr(obj, attr))
            try:
                json.dumps(obj.__dict__.copy())
            except Exception as ex:
                log.exception("Object %s problem" % obj)

        # Resolve associations to real resource IDs
        for refassoc in self.ref_assocs:
            sub_refid, pred, obj_refid = refassoc
            try:
                subo = self.ui_objs[sub_refid]
                objo = self.ui_objs[obj_refid]
                assoc = objects.Association(at="",
                    s=subo._id, st=subo._get_type(), srv="",
                    p=pred,
                    o=objo._id, ot=objo._get_type(), orv="",
                    ts=get_ion_ts())

                self.ui_assocs.append(assoc)
            except Exception as ex:
                log.warn("Cannot create association for subject=%s pred=%s object=%s: %s" % (sub_refid, pred, obj_refid, ex))