# Path: pylm/servers.py
# class Pipeline(Server):
#     """
#     Minimal server that acts as a pipeline.
# 
#     :param str name: Name of the server
#     :param str db_address: ZeroMQ address of the cache service.
#     :param str sub_address: Address of the pub socket of the previous server
#     :param str pub_address: Address of the pub socket
#     :param previous: Name of the previous server.
#     :param to_client: True if the message is sent back to the client. Defaults to True 
#     :param log_level: Minimum output log level.
#     :param int messages: Total number of messages that the server processes.
#         Useful for debugging.
#     """
#     def __init__(self, name, db_address,
#                  sub_address, pub_address, previous, to_client=True,
#                  log_level=logging.INFO, messages=sys.maxsize):
#         self.name = name
#         self.cache = DictDB()
#         self.db_address = db_address
#         self.sub_address = sub_address
#         self.pub_address = pub_address
#         self.pipelined = not to_client
#         self.message = None
# 
#         self.cache.set('name', name.encode('utf-8'))
#         self.cache.set('sub_address', sub_address.encode('utf-8'))
#         self.cache.set('pub_address', pub_address.encode('utf-8'))
# 
#         self.logger = logging.getLogger(name=name)
#         handler = logging.StreamHandler(sys.stdout)
#         handler.setFormatter(
#             logging.Formatter(
#                 '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
#             )
#         )
#         self.logger.addHandler(handler)
#         self.logger.setLevel(log_level)
# 
#         self.messages = messages
# 
#         self.sub_socket = zmq_context.socket(zmq.SUB)
#         self.sub_socket.setsockopt_string(zmq.SUBSCRIBE, previous)
#         self.sub_socket.connect(self.sub_address)
# 
#         self.pub_socket = zmq_context.socket(zmq.PUB)
#         self.pub_socket.bind(self.pub_address)
# 
#     def _execution_handler(self):
#         for i in range(self.messages):
#             self.logger.debug('Server waiting for messages')
#             message_data = self.sub_socket.recv_multipart()[1]
#             self.logger.debug('Got message {}'.format(i + 1))
#             result = b'0'
#             self.message = PalmMessage()
#             try:
#                 self.message.ParseFromString(message_data)
# 
#                 # Handle the fact that the message may be a complete pipeline
#                 try:
#                     if ' ' in self.message.function:
#                         [server, function] = self.message.function.split()[
#                             self.message.stage].split('.')
#                     else:
#                         [server, function] = self.message.function.split('.')
#                 except IndexError:
#                     raise ValueError('Pipeline call not correct. Review the '
#                                      'config in your client')
#                         
#                 if not self.name == server:
#                     self.logger.error('You called {}, instead of {}'.format(
#                         server, self.name))
#                 else:
#                     try:
#                         user_function = getattr(self, function)
#                         self.logger.debug('Looking for {}'.format(function))
#                         try:
#                             result = user_function(self.message.payload)
#                         except:
#                             self.logger.error('User function gave an error')
#                             exc_type, exc_value, exc_traceback = sys.exc_info()
#                             lines = traceback.format_exception(
#                                 exc_type, exc_value, exc_traceback)
#                             for l in lines:
#                                 self.logger.exception(l)
# 
#                     except KeyError:
#                         self.logger.error(
#                             'Function {} was not found'.format(function)
#                         )
#             except DecodeError:
#                 self.logger.error('Message could not be decoded')
# 
#             # Do nothing if the function returns no value
#             if result is None:
#                 continue
# 
#             self.message.payload = result
# 
#             topic, self.message = self.handle_stream(self.message)
#             self.pub_socket.send_multipart(
#                 [topic.encode('utf-8'), self.message.SerializeToString()]
#             )


from pylm.servers import Pipeline


class MyPipeline(Pipeline):
    def foo(self, message):
        self.logger.info('Got a message')
        return b'and I pipelined ' + message


if __name__ == '__main__':
    server = MyPipeline('my_pipeline',
                        db_address='tcp://127.0.0.1:5570',
                        sub_address='tcp://127.0.0.1:5557',

# Path: pylm/servers.py
# class Pipeline(Server):
#     """
#     Minimal server that acts as a pipeline.
# 
#     :param str name: Name of the server
#     :param str db_address: ZeroMQ address of the cache service.
#     :param str sub_address: Address of the pub socket of the previous server
#     :param str pub_address: Address of the pub socket
#     :param previous: Name of the previous server.
#     :param to_client: True if the message is sent back to the client. Defaults to True 
#     :param log_level: Minimum output log level.
#     :param int messages: Total number of messages that the server processes.
#         Useful for debugging.
#     """
#     def __init__(self, name, db_address,
#                  sub_address, pub_address, previous, to_client=True,
#                  log_level=logging.INFO, messages=sys.maxsize):
#         self.name = name
#         self.cache = DictDB()
#         self.db_address = db_address
#         self.sub_address = sub_address
#         self.pub_address = pub_address
#         self.pipelined = not to_client
#         self.message = None
# 
#         self.cache.set('name', name.encode('utf-8'))
#         self.cache.set('sub_address', sub_address.encode('utf-8'))
#         self.cache.set('pub_address', pub_address.encode('utf-8'))
# 
#         self.logger = logging.getLogger(name=name)
#         handler = logging.StreamHandler(sys.stdout)
#         handler.setFormatter(
#             logging.Formatter(
#                 '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
#             )
#         )
#         self.logger.addHandler(handler)
#         self.logger.setLevel(log_level)
# 
#         self.messages = messages
# 
#         self.sub_socket = zmq_context.socket(zmq.SUB)
#         self.sub_socket.setsockopt_string(zmq.SUBSCRIBE, previous)
#         self.sub_socket.connect(self.sub_address)
# 
#         self.pub_socket = zmq_context.socket(zmq.PUB)
#         self.pub_socket.bind(self.pub_address)
# 
#     def _execution_handler(self):
#         for i in range(self.messages):
#             self.logger.debug('Server waiting for messages')
#             message_data = self.sub_socket.recv_multipart()[1]
#             self.logger.debug('Got message {}'.format(i + 1))
#             result = b'0'
#             self.message = PalmMessage()
#             try:
#                 self.message.ParseFromString(message_data)
# 
#                 # Handle the fact that the message may be a complete pipeline
#                 try:
#                     if ' ' in self.message.function:
#                         [server, function] = self.message.function.split()[
#                             self.message.stage].split('.')
#                     else:
#                         [server, function] = self.message.function.split('.')
#                 except IndexError:
#                     raise ValueError('Pipeline call not correct. Review the '
#                                      'config in your client')
#                         
#                 if not self.name == server:
#                     self.logger.error('You called {}, instead of {}'.format(
#                         server, self.name))
#                 else:
#                     try:
#                         user_function = getattr(self, function)
#                         self.logger.debug('Looking for {}'.format(function))
#                         try:
#                             result = user_function(self.message.payload)
#                         except:
#                             self.logger.error('User function gave an error')
#                             exc_type, exc_value, exc_traceback = sys.exc_info()
#                             lines = traceback.format_exception(
#                                 exc_type, exc_value, exc_traceback)
#                             for l in lines:
#                                 self.logger.exception(l)
# 
#                     except KeyError:
#                         self.logger.error(
#                             'Function {} was not found'.format(function)
#                         )
#             except DecodeError:
#                 self.logger.error('Message could not be decoded')
# 
#             # Do nothing if the function returns no value
#             if result is None:
#                 continue
# 
#             self.message.payload = result
# 
#             topic, self.message = self.handle_stream(self.message)
#             self.pub_socket.send_multipart(
#                 [topic.encode('utf-8'), self.message.SerializeToString()]
#             )


# Path: examples/stream_sink/pipeline_odd.py
from pylm.servers import Pipeline



class MyPipeline(Pipeline):
    def foo(self, message):
        self.logger.info('Got a message')
        return b'and I pipelined ' + message


if __name__ == '__main__':
    server = MyPipeline('my_pipeline',
                        db_address='tcp://127.0.0.1:5570',
                        sub_address='tcp://127.0.0.1:5557',

                        pub_address='tcp://127.0.0.1:5571',