Feature/SK-671 | Merge services in combiner, proto refactor

fedn/fedn/network/clients/client.py

@@ -222,7 +222,6 @@ def _connect(self, client_config):
 
         self.channel = channel
 
-        self.connectorStub = rpc.ConnectorStub(channel)
         self.combinerStub = rpc.CombinerStub(channel)
         self.modelStub = rpc.ModelServiceStub(channel)
 
@@ -693,7 +692,7 @@ def _send_heartbeat(self, update_frequency=2.0):
             heartbeat = fedn.Heartbeat(sender=fedn.Client(
                 name=self.name, role=fedn.WORKER))
             try:
-                self.connectorStub.SendHeartbeat(heartbeat, metadata=self.metadata)
+                self.combinerStub.SendHeartbeat(heartbeat, metadata=self.metadata)
                 self._missed_heartbeat = 0
             except grpc.RpcError as e:
                 status_code = e.code()
@@ -733,7 +732,7 @@ def _send_status(self, msg, log_level=fedn.Status.INFO, type=None, request=None)
         self.logs.append(
             "{} {} LOG LEVEL {} MESSAGE {}".format(str(datetime.now()), status.sender.name, status.log_level,
                                                    status.status))
-        _ = self.connectorStub.SendStatus(status, metadata=self.metadata)
+        _ = self.combinerStub.SendStatus(status, metadata=self.metadata)
 
     def run(self):
         """ Run the client. """

fedn/fedn/network/combiner/combiner.py

@@ -50,7 +50,7 @@ def role_to_proto_role(role):
         return fedn.OTHER
 
 
-class Combiner(rpc.CombinerServicer, rpc.ReducerServicer, rpc.ConnectorServicer, rpc.ControlServicer):
+class Combiner(rpc.CombinerServicer):
     """ Combiner gRPC server.
 
     :param config: configuration for the combiner
@@ -400,7 +400,7 @@ def _flush_model_update_queue(self):
 
     # Controller Service
 
-    def Start(self, control: fedn.ControlRequest, context):
+    def StartRound(self, control: fedn.ControlRequest, context):
         """ Start a round of federated learning"
 
         :param control: the control request
@@ -475,7 +475,7 @@ def FlushAggregationQueue(self, control: fedn.ControlRequest, context):
 
     ##############################################################################
 
-    def Stop(self, control: fedn.ControlRequest, context):
+    def StopRound(self, control: fedn.ControlRequest, context):
         """ TODO: Not yet implemented.
 
         :param control: the control request

fedn/fedn/network/combiner/interfaces.py

@@ -175,7 +175,7 @@ def flush_model_update_queue(self):
 
         channel = Channel(self.address, self.port,
                           self.certificate).get_channel()
-        control = rpc.ControlStub(channel)
+        control = rpc.CombinerStub(channel)
 
         request = fedn.ControlRequest()
 
@@ -196,7 +196,7 @@ def set_aggregator(self, aggregator):
 
         channel = Channel(self.address, self.port,
                           self.certificate).get_channel()
-        control = rpc.ControlStub(channel)
+        control = rpc.CombinerStub(channel)
 
         request = fedn.ControlRequest()
         p = request.parameter.add()
@@ -221,7 +221,7 @@ def submit(self, config):
         """
         channel = Channel(self.address, self.port,
                           self.certificate).get_channel()
-        control = rpc.ControlStub(channel)
+        control = rpc.CombinerStub(channel)
         request = fedn.ControlRequest()
         request.command = fedn.Command.START
         for k, v in config.items():
@@ -230,7 +230,7 @@ def submit(self, config):
             p.value = str(v)
 
         try:
-            response = control.Start(request)
+            response = control.StartRound(request)
         except grpc.RpcError as e:
             if e.code() == grpc.StatusCode.UNAVAILABLE:
                 raise CombinerUnavailableError
@@ -272,7 +272,7 @@ def allowing_clients(self):
         """
         channel = Channel(self.address, self.port,
                           self.certificate).get_channel()
-        connector = rpc.ConnectorStub(channel)
+        connector = rpc.CombinerStub(channel)
         request = fedn.ConnectionRequest()
 
         try:
@@ -302,7 +302,7 @@ def list_active_clients(self, queue=1):
         """
         channel = Channel(self.address, self.port,
                           self.certificate).get_channel()
-        control = rpc.ConnectorStub(channel)
+        control = rpc.CombinerStub(channel)
         request = fedn.ListClientsRequest()
         request.channel = queue
         try: