change the publish to request in destroy methods

Author: shohamroditimemphis

memphis/memphis.py

@@ -271,8 +271,12 @@ async def destroy(self):
                 "factory_name":self.name
             }
             factory_name = json.dumps(nameReq, indent=2).encode('utf-8')
-            await self.connection.broker_connection.publish('$memphis_factory_destructions', factory_name)
-
+            res = await self.connection.broker_manager.request('$memphis_factory_destructions', factory_name)
+            error = res.data.decode('utf-8')
+            if "mongo: no documents in result" in error:
+                print('Producer already destroyed')
+            elif error != "":
+                raise Exception(error)
         except Exception as e:
             raise Exception(e)
 
@@ -291,7 +295,12 @@ async def destroy(self):
                 "station_name":self.name
             }
             station_name = json.dumps(nameReq, indent=2).encode('utf-8')
-            await self.connection.broker_connection.publish('$memphis_station_destructions', station_name)
+            res = await self.connection.broker_manager.request('$memphis_station_destructions', station_name)
+            error = res.data.decode('utf-8')
+            if "mongo: no documents in result" in error:
+                print('Producer already destroyed')
+            elif error != "":
+                raise Exception(error)
 
         except Exception as e:
             raise Exception(e)
@@ -328,11 +337,16 @@ async def destroy(self):
         try:
             destroyProducerReq = {
                 "name": self.producer_name,
-                "station_name":self.station_name
+                "station_name": self.station_name
             }
-            producer_name = json.dumps(destroyProducerReq, indent=2).encode('utf-8')
-            await self.connection.broker_connection.publish('$memphis_producer_destructions', producer_name)
-
+            
+            producer_name = json.dumps(destroyProducerReq).encode('utf-8')
+            res = await self.connection.broker_manager.request('$memphis_producer_destructions', producer_name)
+            error = res.data.decode('utf-8')
+            if "mongo: no documents in result" in error:
+                print('Producer already destroyed')
+            elif error != "":
+                raise Exception(error)
         except Exception as e:
             raise Exception(e)
 
@@ -402,7 +416,12 @@ async def destroy(self):
                 "station_name":self.station_name
             }
             consumer_name = json.dumps(destroyConsumerReq, indent=2).encode('utf-8')
-            await self.connection.broker_connection.publish('$memphis_consumer_destructions', consumer_name)
+            res = await self.connection.broker_manager.request('$memphis_consumer_destructions', consumer_name)
+            error = res.data.decode('utf-8')
+            if "mongo: no documents in result" in error:
+                print('Producer already destroyed')
+            elif error != "":
+                raise Exception(error)
 
         except Exception as e:
             raise Exception(e)