test(pipeline): test pipeline load from yaml

gnes/client/base.py

@@ -15,6 +15,7 @@
 
 
 import zmq
+from termcolor import colored
 
 from ..helper import set_logger
 from ..proto import send_message, gnes_pb2, recv_message
@@ -33,7 +34,10 @@ def __init__(self, args):
                                                 getattr(self, 'identity', None))
         self.sender, send_addr = build_socket(self.ctx, self.args.host_out, self.args.port_out, self.args.socket_out,
                                               getattr(self, 'identity', None))
-        self.logger.info('send via %s, receive via %s' % (send_addr, recv_addr))
+        self.logger.info(
+            'input %s:%s\t output %s:%s' % (
+                self.args.host_in, colored(self.args.port_in, 'yellow'),
+                self.args.host_out, colored(self.args.port_out, 'yellow')))
 
     def __enter__(self):
         return self
@@ -47,7 +51,10 @@ def close(self):
         self.ctx.term()
 
     def send_message(self, message: "gnes_pb2.Message", timeout: int = -1):
+        self.logger.info('send message: %s' % message.envelope)
         send_message(self.sender, message, timeout=timeout)
 
     def recv_message(self, timeout: int = -1) -> gnes_pb2.Message:
-        return recv_message(self.receiver, timeout=timeout)
+        r = recv_message(self.receiver, timeout=timeout)
+        self.logger.info('recv a message: %s' % r.envelope)
+        return r

gnes/service/frontend.py

@@ -73,13 +73,12 @@ def add_envelope(self, body: 'gnes_pb2.Request', zmq_client: 'ZmqClient'):
             return msg
 
         def remove_envelope(self, m: 'gnes_pb2.Message'):
-            resp = m.response or m.request
+            resp = m.response
             resp.request_id = m.envelope.request_id
             self.logger.info('unpacking a message and return to client: %s' % router2str(m))
             return resp
 
         def Call(self, request, context):
-            self.logger.info('received a new request')
             with self.zmq_context as zmq_client:
                 zmq_client.send_message(self.add_envelope(request, zmq_client), self.args.timeout)
                 return self.remove_envelope(zmq_client.recv_message(self.args.timeout))