allow retrieving and passing session function metadata to avoid datalayer lookups

Author: paarijaat

FunctionWorker/python/MicroFunctionsAPI.py

@@ -252,14 +252,15 @@ def update_metadata(self, metadata_name, metadata_value, is_privileged_metadata=
 
     # session related API calls
     # only valid if the workflow has at least one session function
-    def send_to_running_function_in_session(self, rgid, message, send_now=False):
+    def send_to_running_function_in_session(self, rgid, message, send_now=False, session_metadata=None):
         '''
         Send a message to a long-running session function instance identified with its id in this session.
 
         Args:
             rgid (string): the running long-running session function instance's id.
             message (*): the message to be sent; can be any python data type (<type 'dict', 'list', 'str', 'int', 'float', or 'NoneType'>).
             send_now (boolean): whether the message should be sent immediately or at the end of current function's execution; default: False.
+            session_metadata (string): (optional) the running long-running session function instance's metadata
 
         Returns:
             None
@@ -280,7 +281,7 @@ def send_to_running_function_in_session(self, rgid, message, send_now=False):
 
         if self._is_session_workflow:
             self._session_utils.send_to_running_function_in_session(
-                rgid, message, send_now)
+                rgid, message, send_now, session_metadata)
         else:
             self._logger.warning(
                 "Cannot send a session update message in a workflow with no session functions.")
@@ -767,6 +768,32 @@ def get_all_session_function_ids(self):
                 "Cannot get session function ids in a workflow with no session functions.")
         return rgidlist
 
+    def get_all_session_metadata(self):
+        '''
+        Retrieve a dict of metadata related to all ids of the session function instances in this session.
+
+        Args:
+            None
+
+        Returns:
+            Dict of id (string) -> metadata (string) of all the session function instances in this session.
+
+        Warns:
+            When the calling function is not part of a workflow with at least one session function.
+
+        Note:
+            The usage of this function is only possible with a KNIX-specific feature (i.e., session functions).
+            Using a KNIX-specific feature might make the workflow description incompatible with other platforms.
+
+        '''
+        session_metadata_dict = {}
+        if self._is_session_workflow:
+            session_metadata_dict = self._session_utils.get_all_session_metadata()
+        else:
+            self._logger.warning(
+                "Cannot get session metadata in a workflow with no session functions.")
+        return session_metadata_dict
+
     def is_still_running(self):
         '''
         Retrieve the status of this session function instance.

FunctionWorker/python/SessionUtils.py

@@ -211,6 +211,13 @@ def get_all_session_function_ids(self):
         rgidlist = list(rgidset)
         return rgidlist
 
+    def get_all_session_metadata(self):
+        session_metadata_dict = self._global_data_layer_client.retrieveMap(self._map_name_session_functions)
+        if session_metadata_dict != None and type(session_metadata_dict) == type({}):
+            return session_metadata_dict
+        else:
+            return {}
+
     def get_all_session_function_aliases(self):
         alias_map = {}
         alias_map = self._global_data_layer_client.retrieveMap(self._map_name_session_function_alias_id)
@@ -427,10 +434,12 @@ def is_session_function_running(self):
     # API to send a message to another session function
     # check the locally running functions, and send them the message locally if so
     # otherwise, send it to the EventGlobalPublisher's queue
-    def send_to_running_function_in_session(self, session_function_id, message, send_now=False):
+    def send_to_running_function_in_session(self, session_function_id, message, send_now=False, session_metadata=None):
         #self._logger.debug("[SessionUtils] Sending message to running function: " + str(session_function_id) + " now: " + str(send_now))
         # send the message to the specific running function id
-        function_metadatastr = self._global_data_layer_client.getMapEntry(self._map_name_session_functions, session_function_id)
+        function_metadatastr = session_metadata
+        if function_metadatastr == None or function_metadatastr == '' or type(function_metadatastr) != type(''):
+            function_metadatastr = self._global_data_layer_client.getMapEntry(self._map_name_session_functions, session_function_id)
         self._logger.debug(f"[SessionUtils] Sending message to running function: {str(session_function_id)}, now: {str(send_now)}, function_metadata: {str(function_metadatastr)}")
         try:
             #self._logger.debug("[SessionUtils] function metadata: " + function_metadatastr)