refactored inbound and outbound data processing in federation client and federation server services, updated version information, fixed bug which shared intraserver private geochats between federates

Author: naman108

FreeTAKServer/controllers/configuration/MainConfig.py

@@ -11,7 +11,7 @@ class MainConfig:
     """
 
     # the version information of the server (recommended to leave as default)
-    version = 'FreeTAKServer-1.9.7.6 Public'
+    version = 'FreeTAKServer-1.9.8 Public'
     #
     yaml_path = str(os.environ.get('FTS_CONFIG_PATH', '/opt/FTSConfig.yaml'))
 

FreeTAKServer/controllers/services/Orchestrator.py

@@ -104,7 +104,7 @@ def clientdatapipe_status_check(self) -> bool:
         else:  # otherwise return true
             return True
 
-    def remove_client_information(self, client_information):
+    def remove_service_user(self, client_information):
         """ this method generates the presence object from the
         client_information parameter and sends it as a remove message
         to the client data pipe
@@ -155,7 +155,7 @@ def update_client_information(self, client_information):
             self.logger.error("exception has been thrown updating client data in queue "+str(e))
             raise e
 
-    def add_client_information(self, client_information):
+    def add_service_user(self, client_information):
         """ this method generates the presence and connection objects from the
         client_information parameter and sends it to
 
@@ -318,7 +318,7 @@ def clientConnected(self, rawConnectionInformation):
             # self.logger.info(loggingConstants.CLIENTCONNECTEDFINISHED + str(clientInformation.modelObject.detail.contact.callsign))
             print("adding client")
             self.clientInformationQueue[clientInformation.modelObject.uid] = [clientInformation.socket]
-            self.add_client_information(client_information=clientInformation)
+            self.add_service_user(client_information=clientInformation)
             self.get_client_information()
             print("client added")
             self.sendUserConnectionGeoChat(clientInformation)
@@ -466,7 +466,7 @@ def clientDisconnected(self, clientInformation: User):
                 'there has been an error in a clients disconnection while adding information to the database ' + str(e))
             pass
         try:
-            self.remove_client_information(client_information=clientInformation)
+            self.remove_service_user(client_information=clientInformation)
             # working
             # time.sleep(1)
             print('stage 1 c')

FreeTAKServer/controllers/services/federation/FederationClientService.py

@@ -9,7 +9,8 @@
 #######################################################
 from FreeTAKServer.controllers.configuration.MainConfig import MainConfig
 from FreeTAKServer.controllers.configuration.types import Types
-from FreeTAKServer.controllers.services.federation.handlers import StopHandler, DisconnectHandler, ConnectHandler, SendDataHandler, SendConnectionDataHandler, SendDisconnectionDataHandler
+from FreeTAKServer.controllers.services.federation.handlers import StopHandler, DisconnectHandler, ConnectHandler, SendDataHandler, SendConnectionDataHandler, SendDisconnectionDataHandler, DestinationValidationHandler, DataValidationHandler, HandlerBase
+from FreeTAKServer.controllers.services.federation.external_data_handlers import *
 from FreeTAKServer.model.protobufModel.fig_pb2 import FederatedEvent
 
 
@@ -21,7 +22,7 @@
 from FreeTAKServer.model.federate import Federate
 import selectors
 import socket
-from typing import Tuple
+from typing import Tuple, Dict, List
 import ssl
 import codecs
 import threading
@@ -30,9 +31,13 @@
 from FreeTAKServer.controllers.serializers.protobuf_serializer import ProtobufSerializer
 from FreeTAKServer.controllers.serializers.xml_serializer import XmlSerializer
 from FreeTAKServer.controllers.XMLCoTController import XMLCoTController
+
+
 from FreeTAKServer.model.SpecificCoT.SendOther import SendOther
 from FreeTAKServer.model.FTSModel.Event import Event
+from FreeTAKServer.model.SpecificCoT.SpecificCoTAbstract import SpecificCoTAbstract
 from FreeTAKServer.model.ClientInformation import ClientInformation
+from FreeTAKServer.model.SQLAlchemy.User import User
 from FreeTAKServer.model.SpecificCoT.SendDisconnect import SendDisconnect
 from FreeTAKServer.controllers.DatabaseControllers.DatabaseController import DatabaseController
 
@@ -50,18 +55,133 @@ class FederationClientServiceController(FederationServiceBase):
 
     def __init__(self):
         self.logger = logger
-        self._define_responsibility_chain()
+        self._define_command_responsibility_chain()
+        self._define_connection_responsibility_chain()
+        self._define_service_responsibility_chain()
+        self._define_external_data_responsibility_chain()
+        self._define_data_responsibility_chain()
         self.pipe = None
         self.federates: {str: Federate} = {}
         self.sel = selectors.DefaultSelector()
+        self.user_dict = {}
+
+    def get_service_users(self) -> List[FederatedEvent]:
+        return self.user_dict.values()
+
+    def add_service_user(self, user: FederatedEvent) -> None:
+        """ add a service user to this services user persistence mechanism
+
+        Returns: None
+
+        """
+        self.user_dict[user.contact.uid] = user
+
+    def remove_service_user(self, user: FederatedEvent):
+        """ remove a service user from this services user persistence mechanism
+
+        Returns: None
+
+        """
+        del self.user_dict[user.contact.uid]
+
+    def define_responsibility_chain(self):
+        pass
 
     def _create_context(self):
         self.context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
         self.context.load_cert_chain(MainConfig.federationCert, MainConfig.federationKey,
                                      password=MainConfig.federationKeyPassword)
         self.context.set_ciphers('DEFAULT@SECLEVEL=1')
 
-    def _define_responsibility_chain(self):
+    def _define_external_data_responsibility_chain(self):
+        """ this method is responsible for defining the responsibility chain which handles external data
+        eg. data sent to FTS by a federate
+
+        Returns:
+
+        """
+        fed_proto_standard_handler = FederationProtobufStandardHandler()
+
+        fed_proto_disconnect_handler = FederationProtobufDisconnectionHandler()
+        fed_proto_disconnect_handler.setNextHandler(fed_proto_standard_handler)
+
+        fed_proto_connection_handler = FederationProtobufConnectionHandler()
+        fed_proto_connection_handler.setNextHandler(fed_proto_disconnect_handler)
+
+        fed_proto_validation_handler = FederationProtobufValidationHandler()
+        fed_proto_validation_handler.setNextHandler(fed_proto_connection_handler)
+
+        self.external_data_chain = fed_proto_validation_handler
+
+    def _call_responsibility_chain(self, command):
+        """ this method is responsible for calling the responsibility chains for all command types:
+            service level commands; start, stop etc
+            Connection level commands; close connection, open connection etc
+            data level commands; send data x, each handler is responsible for some facet of data validation before
+                the connection receives it
+
+        Returns: output from successful handler
+
+        """
+        #if command.level == "SERVICE":
+        if command == "STOP":
+            self.service_chain.Handle(obj = self, command= command)
+
+        # elif command.level == "CONNECTION":
+        elif isinstance(command, tuple) and (command[1] == "DELETE" or command[1] == "CREATE" or command[1] == "UPDATE"):
+            self.connection_chain.Handle(obj=self, command=command)
+
+        #elif command.level == "DATA":
+        if isinstance(command, SpecificCoTAbstract) or isinstance(command, ClientInformation):
+            self.data_chain.Handle(obj = self, command= command)
+
+    def _define_service_responsibility_chain(self):
+        """ this method is responsible for defining the responsibility chain which will handle service level commands;
+            or commands which effect the entire service
+
+        Returns: the entry handler for this responsibility chain
+
+        """
+        stop_handler = StopHandler()
+        self.service_chain = stop_handler
+
+    def _define_connection_responsibility_chain(self):
+        """ this method is responsible for defining the responsibility chain which will handle connection level commands;
+            or commands which effect the status of a connection at the socket level
+
+        Returns: the entry handler for this responsibility chain
+
+        """
+        connect_handler = ConnectHandler()
+        disconnect_handler = DisconnectHandler()
+        disconnect_handler.setNextHandler(connect_handler)
+        self.connection_chain = disconnect_handler
+
+    def _define_data_responsibility_chain(self):
+        """ this method is responsible for defining the responsibility chain which will handle data level commands;
+            or commands which transfer data to a client
+
+        Returns: the entry handler for this responsibility chain
+
+        """
+
+        send_data_handler = SendDataHandler()
+
+        destination_validation_handler = DestinationValidationHandler()
+        destination_validation_handler.setNextHandler(send_data_handler)
+
+        send_disconnection_data_handler = SendDisconnectionDataHandler()
+        send_disconnection_data_handler.setNextHandler(destination_validation_handler)
+
+        send_connection_data_handler = SendConnectionDataHandler()
+        send_connection_data_handler.setNextHandler(send_disconnection_data_handler)
+
+        data_validation_handler = DataValidationHandler()
+        data_validation_handler.setNextHandler(send_connection_data_handler)
+
+        self.data_chain = data_validation_handler
+
+    def _define_command_responsibility_chain(self) -> HandlerBase:
         self.m_StopHandler = StopHandler()
 
         self.m_ConnectHandler = ConnectHandler()
@@ -87,8 +207,13 @@ def main(self):
         outbound_data_thread.start()
         inbound_data_thread.join()
 
+    def serialize_data(self, data_object: FederatedEvent):
+        specific_obj, xmlstring = self._process_protobuff_to_object(data_object)
+        specific_obj.xmlString = etree.tostring(xmlstring)
+        return specific_obj
+
     def outbound_data_handler(self):
-        """ this is the main process respoonsible for receiving data from federates and sharing
+        """ this is the main process responsible for receiving data from federates and sharing
         with FTS core
 
         Returns:
@@ -108,13 +233,10 @@ def outbound_data_handler(self):
                         # event = etree.Element('event')
                         # SpecificCoTObj = XMLCoTController().categorize_type(protobuf_object.type)
                         try:
-                            specific_obj, xmlstring = self._process_protobuff_to_object(protobuf_object)
-                            # specific_obj.xmlString = etree.tostring(xmlstring)
-                            print(etree.tostring(xmlstring))
-                            specific_obj.xmlString = etree.tostring(xmlstring)
-                            self.pipe.put(specific_obj)
+                            serialized_data = self.serialize_data(protobuf_object)
+                            self.send_command_to_core(serialized_data)
                         except Exception as e:
-                            pass
+                            self.logger.warning("there has been an exception thrown in the outbound_data_handler "+str(e))
                         """if isinstance(SpecificCoTObj, SendOtherController):
                             detail = protobuf_object.event.other
                             protobuf_object.event.other = ''
@@ -130,6 +252,11 @@ def outbound_data_handler(self):
             else:
                 time.sleep(MainConfig.MainLoopDelay / 1000)
 
+    def send_command_to_core(self, serialized_data):
+        if self.pipe.sender_queue.full():
+            print('queue full !!!')
+        self.pipe.put(serialized_data)
+
     def inbound_data_handler(self):
         """this is the main process responsible for receiving data from FTS core
 
@@ -141,11 +268,10 @@ def inbound_data_handler(self):
                 command = self.pipe.get()
                 if command:
                     try:
-                        self.m_SendConnectionHandler.Handle(self, command)
+                        self._call_responsibility_chain(command)
                     except Exception as e:
                         pass
-                else:
-                    pass
+
             except Exception as e:
                 self.logger.error(str(e))
 
@@ -201,6 +327,7 @@ def receive_data_from_federate(self, timeout):
                         print(raw_protobuf_message)
                         protobuf_object = FederatedEvent()
                         protobuf_object.ParseFromString(raw_protobuf_message)
+                        self.external_data_chain.Handle(self, protobuf_object)
                         dataarray.append(protobuf_object)
                     except Exception as e:
                         conn.recv(10000)

FreeTAKServer/controllers/services/federation/external_data_handlers.py

@@ -0,0 +1,46 @@
+from FreeTAKServer.controllers.services.federation.handlers import HandlerBase
+
+from FreeTAKServer.model.protobufModel.fig_pb2 import FederatedEvent
+
+class FederationProtobufValidationHandler(HandlerBase):
+    """
+    this handler is responsible for validation that a passed command has the proper type for subsequent handlers
+    """
+    def Handle(self, obj, command):
+
+        if isinstance(command, FederatedEvent):
+            self.callNextHandler(obj, command)
+
+        else:
+            raise TypeError("this command chain only accepts Protobuf data of type FederatedEvent")
+
+class FederationProtobufConnectionHandler(HandlerBase):
+    """
+    this handler is responsible for parsing connection data coming from a federate and adding it to the calling services user dictionary
+    """
+    def Handle(self, obj, command):
+        if command.contact.operation == 1:
+            obj.add_service_user(command)
+            self.callNextHandler(obj, command)
+        else:
+            self.callNextHandler(obj,command)
+
+class FederationProtobufDisconnectionHandler(HandlerBase):
+    """
+    this handler is responsible for parsing clients disconnection data coming from a federate
+    """
+    def Handle(self, obj, command):
+        if command.contact.operation == 4:
+            obj.remove_service_user(command)
+            self.callNextHandler(obj, command)
+        else:
+            self.callNextHandler(obj, command)
+
+class FederationProtobufStandardHandler(HandlerBase):
+    """
+    this handler is responsible for serializing client data and sharing it with FTS core
+    """
+    def Handle(self, obj, command):
+        pass
+        # serialized_data = obj.serialize_data(command)
+        # obj.send_command_to_core(serialized_data)