+ @asyncio.coroutine
+ def _terminate_nsr(self, nsr_id):
+ self._log.debug("NSR {} being terminated".format(nsr_id))
+ jobs = self.cfgm.get_job(nsr_id)
+ for job in jobs:
+ path = CfgAgentJobDtsHandler.cfg_job_xpath(nsr_id, job.id)
+ with self._dts.transaction() as xact:
+ self._log.debug("Deleting job: {}".format(path))
+ job.regh.delete_element(path)
+ self._log.debug("Deleted job: {}".format(path))
+
+ # Remove the NSR id in manager
+ self.cfgm.del_nsr(nsr_id)
+
+ @property
+ def nsr_xpath(self):
+ return "D,/nsr:ns-instance-opdata/nsr:nsr"
+
+ @asyncio.coroutine
+ def register_for_nsr(self):
+ """ Register for NSR changes """
+
+ @asyncio.coroutine
+ def on_prepare(xact_info, query_action, ks_path, msg):
+ """ This NSR is created """
+ self._log.debug("Received NSR instantiate on_prepare (%s:%s:%s)",
+ query_action,
+ ks_path,
+ msg)
+
+ if (query_action == rwdts.QueryAction.UPDATE or
+ query_action == rwdts.QueryAction.CREATE):
+ pass
+ elif query_action == rwdts.QueryAction.DELETE:
+ nsr_id = msg.ns_instance_config_ref
+ asyncio.ensure_future(self._terminate_nsr(nsr_id), loop=self._loop)
+ else:
+ raise NotImplementedError(
+ "%s action on cm-state not supported",
+ query_action)
+
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK)
+
+ try:
+ handler = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare)
+ self._nsr_regh = yield from self._dts.register(self.nsr_xpath,
+ flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
+ handler=handler)
+ except Exception as e:
+ self._log.error("Failed to register for NSR changes as %s", str(e))
+