添加进程通信消息类

MingMingWu · MingMingWu · commit b1a1e4717967 · 2017-08-15T02:32:36.000+08:00
diff --git a/multi_thread.py b/multi_thread.py
@@ -61,13 +61,124 @@ def start_thread():
     thread2.start()
 
 
+class Message():
+    """class"""
+    def __init__(self):
+        self.s_pid = 0
+        self.s_pn = ""
+        self.d_pid = 0
+        self.d_pn = ""
+        self.method = "PUT"
+        self.command = 0
+        self.payload = None
+
+    def get_payload(self):
+        return self.payload
+
+    def get_source(self):
+        return self.s_pn, self.s_pid
+
+    def get_destination(self):
+        return self.d_pn, self.d_pid
+
+    def get_method(self):
+        return self.method
+
+    def get_command(self):
+        return self.command
+
+
+class MessageProcess():
+    """process message"""
+
+    def __init__(self, process_name, process_id, queue_in, queue_out, call_back):
+        """init"""
+        self.process_id = process_id
+        self.process_name = process_name
+        self.msg_queue_in = queue_in
+        self.msg_queue_out = queue_out
+        self.msg_in_count = 0
+        self.msg_out_count = 0
+        # 与本进程相连的其他进程，只能和列表中的进程相通信
+        self.process_link_info = {}
+
+        # 回调函数，在调用获取消息接口后会调用回调函数进行处理
+        self.call_back = call_back
+
+        self.message_send = Message()
+        self.message_receive = Message()
+
+    def add_process_link(self, process_name, process_id, queue_in, queue_out):
+        """add process link"""
+
+        if process_name in self.process_link_info:
+            return False, "process_name in link"
+
+        for pro in self.process_link_info:
+            if self.process_link_info[pro]['process_id'] == process_id:
+                return False, "process_id in link"
+
+        new_link = None
+        new_link[process_name] = {
+            'process_id': process_id,
+            'queue_in': queue_in,
+            'queue_out': queue_out,
+            'msg_send_count': 0,
+            'msg_receive_count': 0
+        }
+
+        self.process_link_info[process_name] = new_link
+        return True, "process link add success"
+
+    def del_process_link(self, *agr):
+
+        if isinstance(type(agr), type(1)) is True:
+            for pro in self.process_link_info:
+                if agr == self.process_link_info[pro]['process_id']:
+                    del self.process_link_info[pro]
+        if isinstance(type(agr), type("")) is True:
+            if agr in self.process_link_info:
+                del self.process_link_info[agr]
+        return True
+
+    def send_msg(self, process_name, method, command, data):
+        """send msg"""
+        for pro in self.process_link_info:
+            if process_name == self.process_link_info:
+                self.message_send.d_pid = self.process_link_info[pro]
+                self.message_send.d_pn = process_name
+                self.message_send.s_pid = self.process_id
+                self.message_send.s_pn = self.process_name
+                self.message_send.method = method
+                self.message_send.command = command
+                self.message_send.payload = data
+
+                # 发送数据
+                self.process_link_info[pro]['msg_in'].put(self.message_send)
+                self.msg_out_count += 1
+                self.process_link_info[pro]['msg_send'] += 1
+            else:
+                return False, "process not in link"
+
+    def get_msg(self):
+        if self.msg_queue_in.empty() is True:
+            return False, "there is no message"
+        else:
+            new_msg = self.msg_queue_in.get()
+            self.message_receive.method = new_msg['method']
+            self.message_receive.s_pn = new_msg['s_pn']
+            self.message_receive.s_pid = new_msg['s_pid']
+            self.message_receive.d_pn = new_msg['d_pn']
+            self.message_receive.d_pid = new_msg['d_pid']
+            self.message_receive.command = new_msg['command']
+            self.message_receive.payload = new_msg['payload']
+
+
 def my_process(process_name, queue_up, queue_down):
-    """my process"""
 
     pid = os.getpid()
     print "process_name:%s, process_id:%d" % (process_name, pid)
     queue_up.put(pid)
-
     while True:
         # 进程接收和处理从其他进程发过来的消息
         if queue_down.empty() is True:
@@ -101,7 +212,7 @@ def my_process(process_name, queue_up, queue_down):
     # b = my_thread()
 
     msg_channal = {}
-    for count in range(0, 5, 1):
+    for count in range(0, 1, 1):
         # 创建进程和通信通道
         msg_queue = {}
         queue_up = Queue()
