Modify message processing logic

L2/dockerfiles/admL2/admL2.py

@@ -10,32 +10,45 @@ class admL2Api():
         self.a = 0
     
     def sum_all(self,body):
-        
+
         #取brickid,name1,name2,name3中brickid
         brickid=body.split(",",1)[0]
         #取brickid后面的name1,name2,name3
         body=body.split(",",1)[1]
         admL2Api.sum_numbers(body,brickid)
+        
 
     @classmethod 
     def sum_numbers(self,body,brickid):
-        #取name1,name2,name3中第一个逗号之前字符
-        name=body.split(",",1)[0]
-        #取name1,name2,name3中第一个逗号之后字符
-        body=body.split(",",1)[1]
         #取环境变量中模块id,需预先设定
         current_job_id=os.getenv('CSST_ADML2_JOB_ID')
-
         #调用grpc的SendJobMessage
         channel = grpc.insecure_channel(os.getenv('CSST_PIPELINE_GRPC_SERVER')) 
+
+        if "," in body:
+            #取name1,name2,name3中第一个逗号之前字符
+            name=body.split(",",1)[0]
+            #取name1,name2,name3中第一个逗号之后字符
+            body=body.split(",",1)[1]
+            stub = control_pb2_grpc.ControlServiceStub(channel)
+            test = control_pb2.JobKey()
+            test.builtin_job_id.current_job_id = int(current_job_id)
+            test.builtin_job_id.sink_job_name = name
+            test.key_text = brickid+","+body
+            reflag = stub.SendJobMessage(test)
+            print(reflag.value)
+            return "执行完毕1"
+
         stub = control_pb2_grpc.ControlServiceStub(channel)
         test = control_pb2.JobKey()
         test.builtin_job_id.current_job_id = int(current_job_id)
-        test.builtin_job_id.sink_job_name = name
-        test.key_text = brickid+","+body
+        test.builtin_job_id.sink_job_name = body
+        test.key_text = brickid
         reflag = stub.SendJobMessage(test)
         print(reflag.value)
         return "执行完毕"
+        
+
 if __name__ == '__main__':
     parameter = sys.argv
     body=parameter[1]

L2/dockerfiles/admL2/test.py

-import os
-os.environ['CSST_PIPELINE_GRPC_SERVER']="10.255.2.11:50051"
-os.environ['CSST_ADML2_JOB_ID']="2"
-os.environ['current_job_id']="1"
-from admL2 import admL2Api
-body="brikd,name1,admL2"
-admL2Api().sum_all(body)