X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=blobdiff_plain;f=osm_nbi%2Fsubscriptions.py;h=557bf800f0915d4fa512c39e92396750d7bf7ae9;hp=8132fe9e7329f96a0fe9738c84f7da2933a6f03c;hb=c768937d5bb7b31e827aca36c9ab8c3cf1fdb34b;hpb=09e6542ceae8ee08ee7cc372f7344a77bc1f5304 diff --git a/osm_nbi/subscriptions.py b/osm_nbi/subscriptions.py index 8132fe9..557bf80 100644 --- a/osm_nbi/subscriptions.py +++ b/osm_nbi/subscriptions.py @@ -48,7 +48,7 @@ class SubscriptionThread(threading.Thread): :param engine: an instance of Engine class, used for deleting instances """ threading.Thread.__init__(self) - + self.to_terminate = False self.config = config self.db = None self.msg = None @@ -57,10 +57,12 @@ class SubscriptionThread(threading.Thread): self.logger = logging.getLogger("nbi.subscriptions") self.aiomain_task = None # asyncio task for receiving kafka bus self.internal_session = { # used for a session to the engine methods - "_id": "subscription", - "id": "subscription", - "project_id": "admin", - "admin": True + "project_id": (), + "set_project": (), + "admin": True, + "force": False, + "public": None, + "method": "delete", } def run(self): @@ -97,16 +99,17 @@ class SubscriptionThread(threading.Thread): raise SubscriptionException(str(e), http_code=e.http_code) self.logger.debug("Starting") - while True: + while not self.to_terminate: try: self.aiomain_task = asyncio.ensure_future(self.msg.aioread(("ns", "nsi"), loop=self.loop, callback=self._msg_callback), loop=self.loop) self.loop.run_until_complete(self.aiomain_task) - except asyncio.CancelledError: - break # if cancelled it should end, breaking loop + # except asyncio.CancelledError: + # break # if cancelled it should end, breaking loop except Exception as e: - self.logger.exception("Exception '{}' at messaging read loop".format(e), exc_info=True) + if not self.to_terminate: + self.logger.exception("Exception '{}' at messaging read loop".format(e), exc_info=True) self.logger.debug("Finishing") self._stop() @@ -160,4 +163,5 @@ class SubscriptionThread(threading.Thread): but not immediately. :return: None """ + self.to_terminate = True self.loop.call_soon_threadsafe(self.aiomain_task.cancel)