Add tenant name to requests (#148)

* add tenant name to requests

* update the timeout to 2 ms

Author: shohamroditimemphis

memphis/consumer.py

@@ -45,6 +45,7 @@ def __init__(
         self.dls_current_index = 0
         self.dls_callback_func = None
         self.t_dls = asyncio.create_task(self.__consume_dls())
+        self.t_consume = None
 
     def set_context(self, context):
         """Set a context (dict) that will be passed to each message handler call."""
@@ -178,6 +179,7 @@ async def destroy(self):
                 "name": self.consumer_name,
                 "station_name": self.station_name,
                 "username": self.connection.username,
+                "tenant_name": self.connection.tenant_name
             }
             consumer_name = json.dumps(destroyConsumerReq, indent=2).encode("utf-8")
             res = await self.connection.broker_manager.request(

memphis/memphis.py

@@ -128,7 +128,7 @@ async def connect(
         reconnect: bool = True,
         max_reconnect: int = 10,
         reconnect_interval_ms: int = 1500,
-        timeout_ms: int = 15000,
+        timeout_ms: int = 2000,
         cert_file: str = "",
         key_file: str = "",
         ca_file: str = "",
@@ -248,6 +248,7 @@ async def station(
                 },
                 "username": self.username,
                 "tiered_storage_enabled": tiered_storage_enabled,
+                "tenant_name": self.tenant_name
             }
             create_station_req_bytes = json.dumps(createStationReq, indent=2).encode(
                 "utf-8"
@@ -278,7 +279,7 @@ async def attach_schema(self, name, stationName):
         try:
             if name == "" or stationName == "":
                 raise MemphisError("name and station name can not be empty")
-            msg = {"name": name, "station_name": stationName, "username": self.username}
+            msg = {"name": name, "station_name": stationName, "username": self.username, "tenant_name": self.tenant_name}
             msgToSend = json.dumps(msg).encode("utf-8")
             err_msg = await self.broker_manager.request(
                 "$memphis_schema_attachments", msgToSend, timeout=5
@@ -300,7 +301,7 @@ async def detach_schema(self, stationName):
         try:
             if stationName == "":
                 raise MemphisError("station name is missing")
-            msg = {"station_name": stationName, "username": self.username}
+            msg = {"station_name": stationName, "username": self.username, "tenant_name": self.tenant_name}
             msgToSend = json.dumps(msg).encode("utf-8")
             err_msg = await self.broker_manager.request(
                 "$memphis_schema_detachments", msgToSend, timeout=5
@@ -385,6 +386,7 @@ async def producer(
                 "producer_type": "application",
                 "req_version": 1,
                 "username": self.username,
+                "tenant_name": self.tenant_name
             }
             create_producer_req_bytes = json.dumps(createProducerReq, indent=2).encode(
                 "utf-8"
@@ -417,7 +419,7 @@ async def producer(
                 if self.schema_updates_data[internal_station_name]["type"] == "json":
                     schema = self.schema_updates_data[internal_station_name][
                         "active_version"
-                    ]["schema_content"]
+    ]["schema_content"]
                     self.json_schemas[internal_station_name] = json.loads(schema)
                 elif (
                     self.schema_updates_data[internal_station_name]["type"] == "graphql"
@@ -562,6 +564,7 @@ async def consumer(
                 "last_messages": last_messages,
                 "req_version": 1,
                 "username": self.username,
+                "tenant_name": self.tenant_name
             }
 
             create_consumer_req_bytes = json.dumps(createConsumerReq, indent=2).encode(

memphis/producer.py

@@ -273,6 +273,7 @@ async def destroy(self):
                 "name": self.producer_name,
                 "station_name": self.station_name,
                 "username": self.connection.username,
+                "tenant_name": self.connection.tenant_name
             }
 
             producer_name = json.dumps(destroyProducerReq).encode("utf-8")

memphis/station.py

@@ -12,7 +12,7 @@ def __init__(self, connection, name: str):
     async def destroy(self):
         """Destroy the station."""
         try:
-            nameReq = {"station_name": self.name, "username": self.connection.username}
+            nameReq = {"station_name": self.name, "username": self.connection.username, "tenant_name": self.connection.tenant_name}
             station_name = json.dumps(nameReq, indent=2).encode("utf-8")
             res = await self.connection.broker_manager.request(
                 "$memphis_station_destructions", station_name, timeout=5