class Dropped(Exception):
class Rejected(Exception):
class ChannelStopped(Exception):
class BaseChannel:
class SubChannel(BaseChannel):
class ConditionSubChannel(BaseChannel):
class Case():
class FileWatcherChannel(BaseChannel):
    STARTING, WAITING, PROCESSING, STOPPING, STOPPED = range(5)
    STATE_NAMES = ['STARTING', 'WAITING', 'PROCESSING', 'STOPPING', 'STOPPED']
    NEW, UNCHANGED, MODIFIED, DELETED = range(4)
    def __init__(self, name=None, parent_channel=None, loop=None, message_store_factory=None):
    def status_id_to_str(cls, state_id):
    def status_str_to_id(cls, state):
    def status(self):
    def status(self, value):
    def is_stopped(self):
    async def start(self):
    def init_node_graph(self):
    async def stop(self):
    def _reset_test(self):
    def add(self, *args):
    def _register_node(self, node):
    def get_node(self, name):
    def append(self, *args):
    def fork(self, name=None, message_store_factory=None):
    def when(self, condition, name=None, message_store_factory=None):
    def case(self, *conditions, names=None, message_store_factory=None):
    def handle_and_wait(self, msg):
    async def handle(self, msg):
    async def subhandle(self, msg):
    async def process(self, msg):
    async def replay(self, msg_id):
    def to_dict(self):
    def subchannels(self):
    def graph(self, prefix='', dot=False):
    def graph_dot(self, end=''):
    def __str__(self):
    def __repr__(self):
def reset_pypeman_channels():
    def callback(self, fut):
    async def process(self, msg):
    def __init__(self, condition=lambda x: True, **kwargs):
    def test_condition(self, msg):
    async def subhandle(self, msg):
    def __init__(self, *args, names=None, parent_channel=None, message_store_factory=None, loop=None):
    def _reset_test(self):
    def test_condition(self, condition, msg):
    async def handle(self, msg):
    def __init__(self, *args, basedir='', regex='.*', interval=1, binary_file=False, path='', **kwargs):
    async def start(self):
    def file_status(self, filename):
    def _handle_callback(self, future):
    async def watch_for_file(self): def reset_pypeman_endpoints():
    def __init__(self):
    async def start(self):
    def __init__(self, loop=None, sock=None, default_port='8080', reuse_port=None):
    def normalize_socket(sock, default_port="8080"):
    def mk_socket(sock, reuse_port):
    def make_socket(self):
class BaseEndpoint:
class SocketEndpoint(BaseEndpoint):
                SO_REUSEPORT = 15 SENTINEL = object()
    NONE = object()
def choose_first_not_none(*args):
def callable_or_value(val, msg):
def get_context(msg, date=None, counter=None):
def find_node(name="", match=None):
    def __init__(self, *args, name=None, log_output=False, **kwargs):
    def fullpath(self):
    async def handle(self, msg):
    async def _log_handle(self, msg):
    async def _test_handle(self, msg):
    async def async_run(self, msg):
    def run(self, msg):
    def process(self, msg):
    async def save_data(self, key, value):
    async def restore_data(self, key, default=SENTINEL):
    def mock(self, input=None, output=None):
            def new_process(msg):
    def _reset_test(self):
    def last_input(self):
    def __str__(self):
    def process(self, msg):
    def __init__(self, message=None, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, ctx_name, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, thread_pool=None, **kwargs):
    def run(self, msg):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, duration=1, **kwargs):
    async def process(self, msg):
    def __init__(self, *args, encoding='utf-8', **kwargs):
    def process(self, msg):
    def __init__(self, *args, encoding='utf-8', indent=None, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, altchars=None, **kwargs):
    def process(self, msg):
    def __init__(self, *args, altchars=None, **kwargs):
    def process(self, msg):
    def store(self, message):
    def __init__(self, path, filename, channel):
    def store(self, message):
    def __init__(self, *args, uri=None, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def __init__(self, filename=None, filepath=None, binary_file=False, *args, **kwargs):
    def process(self, msg):
    def __init__(self, filepath=None, binary_mode=False, safe_file=True, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, **kwargs):
    def __init__(self, *args, **kwargs):
    def process(self, msg):
    def __init__(self, *args, host=None, port=None, user=None, password=None, ssl=False, start_tls=False,
                 subject=None, sender=None, recipients=None, content=None, **kwargs):
    def send_email(self, subject, sender, recipients, content):
    def process(self, msg):
def reset_pypeman_nodes():
class NodeException(Exception):
class BaseNode:
class RaiseError(BaseNode):
class Drop(BaseNode):
class DropNode(Drop):
class Empty(BaseNode):
class SetCtx(BaseNode):
class ThreadNode(BaseNode):
class Log(BaseNode):
class Sleep(BaseNode):
class JsonToPython(BaseNode):
class PythonToJson(BaseNode):
class Encode(BaseNode):
class Decode(BaseNode):
class B64Encode(BaseNode):
class B64Decode(BaseNode):
class SaveNullBackend():
class SaveFileBackend():
class Save(ThreadNode):
class MessageStore(Save):
class FileReader(BaseNode):
class FileWriter(BaseNode):
class Map(BaseNode):
class MappingNode(Map):
class ToOrderedDict(BaseNode):
class Email(ThreadNode): class Event:
    def __init__(self):
    def add_handler(self, handler):
    def remove_handler(self, handler):
    def receiver(self, handler):
    async def fire(self, *args, **kargs):
    def getHandlerCount(self): class BaseChannel:
    """
    Base channel are generic channels.
    If you want to create new channel, inherit from the base class and call
    ``self.handle(msg)`` method with generated message.

    :param name: Channel name is mandatory and must be unique through the whole project.
        Name gives a way to get channel in test mode.

    :param parent_channel: Used with sub channels. Don't specify yourself.

    :param loop: To specify a custom event loop.

    :param message_store_factory:     You can specify a message store (see below) at channel
        initialisation if you want to save all processed message. Use
        `message_store_factory` argument with  an instance of wanted message store factory.
    """
    STARTING, WAITING, PROCESSING, STOPPING, STOPPED = range(5)
    STATE_NAMES = ['STARTING', 'WAITING', 'PROCESSING', 'STOPPING', 'STOPPED']

    def __init__(self, name=None, parent_channel=None, loop=None, message_store_factory=None):

        self.uuid = uuid.uuid4()

        all_channels.append(self)
        self._nodes = []
        self._node_map = {}
        self._status = BaseChannel.STOPPED
        self.processed_msgs = 0
        self.interruptable_sleeper = Sleeper(loop)  # for interruptable sleeps

        if name:
            self.name = name
        else:
            warnings.warn("Channels without names are deprecated", DeprecationWarning)
            self.name = self.__class__.__name__ + "_" + str(len(all_channels))

        self.parent = None
        if parent_channel:
            # Use dot name hierarchy
            self.name = ".".join([parent_channel.name, self.name])

            self.parent = parent_channel

            # TODO parent channels usefull ?
            self.parent_uids = [parent_channel.uuid]
            self.parent_names = [parent_channel.name]
            if parent_channel.parent_uids:
                self.parent_uids.append(parent_channel.parent_uids)
                self.parent_names += parent_channel.parent_names
        else:
            self.parent_uids = None

        if self.name in _channel_names:
            raise NameError(
                "Duplicate channel name %r . "
                "Channel names must be unique !" % self.name
            )

        _channel_names.add(self.name)

        if loop is None:
            self.loop = asyncio.get_event_loop()
        else:
            self.loop = loop

        self.logger = logging.getLogger('pypeman.channels.%s' % self.name)

        self.next_node = None

        self.message_store_factory = message_store_factory or msgstore.NullMessageStoreFactory()

        self.message_store = self.message_store_factory.get_store(self.name)

        self._first_start = True

        # Used to avoid multiple messages processing at same time
        self.lock = asyncio.Lock(loop=self.loop)

    @classmethod
    def status_id_to_str(cls, state_id):
        return cls.STATE_NAMES[state_id]

    @classmethod
    def status_str_to_id(cls, state):
        return cls.STATE_NAMES.index(state)

    @property
    def status(self):
        """ Getter for status """
        return self._status

    @status.setter
    def status(self, value):
        old_state = self._status
        self._status = value
        # Launch change state event
        ensure_future(events.channel_change_state.fire(channel=self, old_state=old_state, new_state=value),
                      loop=self.loop)

    def is_stopped(self):
        """
        :return: True if channel is in stopped or stopping state.
        """
        return self.status in (BaseChannel.STOPPING, BaseChannel.STOPPED,)

    async def start(self):
        """
        Start the channel. Called before starting processus. Can be overloaded to specify specific
        start procedure.
        """
        self.status = BaseChannel.STARTING
        if self._first_start:
            self.init_node_graph()
            self._first_start = False
        await self.message_store.start()
        self.status = BaseChannel.WAITING

    def init_node_graph(self):
        if self._nodes:
            previous_node = self._nodes[0]

            for node in self._nodes[1:]:
                previous_node.next_node = node
                previous_node = node

    async def stop(self):
        """
        Stop the channel.
        Called when
        - pypeman shuts down.
        - a channel is stopped (e.g. via the admin interface)
        """
        self.status = BaseChannel.STOPPING
        # Verify that all messages are processed
        with (await self.lock):
            self.status = BaseChannel.STOPPED
        # stop all pending sleeps
        await self.interruptable_sleeper.cancel_all()

    def _reset_test(self):
        """ Enable test mode and reset node data.
        """
        for node in self._nodes:
            node._reset_test()

    def add(self, *args):
        """
        Add specified nodes to channel (Shortcut for append).

        :param args: Nodes to add.
        """
        return self.append(*args)

    def _register_node(self, node):
        """
        Node registering to search.

        :param node: node to register.
        """
        self._node_map[node.name] = node
        if self.parent:
            self.parent._register_node(node)

    def get_node(self, name):
        """
        Return node with name in argument. Mainly used in tests.

        :param name: The searched node name.

        :return: Instance of Node or None if none found.
        """

        return self._node_map.get(name)

    def append(self, *args):
        """
        Append specified nodes to channel.

        :param args: Nodes to add. can be a list, a tupe or a nested
                list of tuples or lists if nodes.
                nodes with value None are ignored
        """
        args = [node for node in flatten(args) if node]
        for node in args:
            node.channel = self
            self._nodes.append(node)
            self._register_node(node)

        return self

    def fork(self, name=None, message_store_factory=None):
        """
        Create a new channel that processes a copy of the message
        at this point.
        Subchannels are executed in parallel of main process.

        :return: The forked channel
        """

        s = SubChannel(
            name=name, parent_channel=self,
            message_store_factory=message_store_factory, loop=self.loop)
        self._nodes.append(s)
        return s

    def when(self, condition, name=None, message_store_factory=None):
        """
        New channel bifurcation that is executed only if condition is True. This channel
        replace further current channel processing.

        :param condition: Can be a value or a function that takes a message argument.

        :return: The conditional path channel.
        """

        s = ConditionSubChannel(
            condition=condition, name=name, parent_channel=self,
            message_store_factory=message_store_factory, loop=self.loop)
        self._nodes.append(s)
        return s

    def case(self, *conditions, names=None, message_store_factory=None):
        """
        Case between multiple conditions. For each condition specified, a
        channel is returned by this method in same order as condition are
        given.
        When processing a message, conditions are evaluated successively and
        the first returning true triggers the corresponding channel processing
        the message.
        When channel processing is finished, the next node is called.

        :param conditions: Multiple conditions, one for each returned channel.
            Should be a boolean or a function that takes a ``msg`` argument
            and should return a boolean.

        :param message_store_factory: Allows you to specify a message store
            factory for all channel of this `case`.

        :return: one channel by condition parameter.
        """
        if names is None:
            names = [None] * len(conditions)

        c = Case(*conditions, names=names, parent_channel=self,
                 message_store_factory=message_store_factory, loop=self.loop)
        self._nodes.append(c)
        return [chan for cond, chan in c.cases]

    def handle_and_wait(self, msg):
        """ Handle a message synchronously. Mainly used for testing purpose.

        :param msg: Message to process
        :return: Processed message.
        """
        return self.loop.run_until_complete(self.handle(msg))

    async def handle(self, msg):
        """ Overload this method only if you know what you are doing but
        call it from child class to add behaviour.

        :param msg: To be processed msg.
        :return: Processed message
        """

        # Store message before any processing
        # TODO If store fails, do we stop processing ?
        # TODO Do we store message even if channel is stopped ?
        msg_store_id = await self.message_store.store(msg)

        if self.status in [BaseChannel.STOPPED, BaseChannel.STOPPING]:
            raise ChannelStopped("Channel is stopped so you can't send message.")

        self.logger.info("%s handle %s", self, msg)

        # Only one message processing at time
        with (await self.lock):
            self.status = BaseChannel.PROCESSING
            try:
                result = await self.subhandle(msg)
                await self.message_store.change_message_state(msg_store_id, message.Message.PROCESSED)
                return result
            except Dropped:
                await self.message_store.change_message_state(msg_store_id, message.Message.PROCESSED)
                raise
            except Rejected:
                await self.message_store.change_message_state(msg_store_id, message.Message.REJECTED)
                raise
            except Exception:
                self.logger.exception('Error while processing message %s', msg)
                await self.message_store.change_message_state(msg_store_id, message.Message.ERROR)
                raise
            finally:
                self.status = BaseChannel.WAITING
                self.processed_msgs += 1

    async def subhandle(self, msg):
        """ Overload this method only if you know what you are doing. Called by ``handle`` method.

        :param msg: To be processed msg.

        :return: Processed message
        """

        result = await self.process(msg)

        if self.next_node:
            if isinstance(result, types.GeneratorType):
                gene = result
                result = msg  # Necessary if all nodes result are dropped
                for res in gene:
                    try:
                        result = await self.next_node.handle(res)
                    except Dropped:
                        pass
                    # TODO Here result is last value returned. Is it a good idea ?
            else:
                result = await self.next_node.handle(result)

        return result

    async def process(self, msg):
        """ Overload this method only if you know what you are doing. Called by ``subhandle`` method.

        :param msg: To be processed msg.

        :return: Processed message
        """

        if self._nodes:
            res = await self._nodes[0].handle(msg)
            return res
        else:
            return msg

    async def replay(self, msg_id):
        """
        This method allows you to replay a message from channel `message_store`.

        :param msg_id: Message id to replay.

        :return: The result of the processing.
        """
        msg_dict = await self.message_store.get(msg_id)
        new_message = msg_dict['message'].renew()
        result = await self.handle(new_message)
        return result

    def to_dict(self):
        return {
            'name': self.name,
            'status': BaseChannel.status_id_to_str(self.status),
            'has_message_store': not isinstance(self.message_store, msgstore.NullMessageStore),
            'processed': self.processed_msgs,
        }

    def subchannels(self):
        res = []

        for node in self._nodes:
            if isinstance(node, SubChannel) or isinstance(node, ConditionSubChannel):
                chan_dict = node.to_dict()
                chan_dict['subchannels'] = node.subchannels()
                res.append(chan_dict)
            elif isinstance(node, Case):
                for cond, channel in node.cases:
                    chan_dict = channel.to_dict()
                    chan_dict['subchannels'] = channel.subchannels()
                    res.append(chan_dict)
        return res

    def graph(self, prefix='', dot=False):
        """
        Generate a text graph for this channel.
        """
        for node in self._nodes:
            if isinstance(node, SubChannel):
                yield prefix + '|—\\ (%s)' % node.name
                for entry in node.graph(prefix='|  ' + prefix):
                    yield entry
            elif isinstance(node, ConditionSubChannel):
                yield prefix + '|?\\ (%s)' % node.name
                for entry in node.graph(prefix='|  ' + prefix):
                    yield entry
                yield prefix + '|  -> Out'
            elif isinstance(node, Case):
                for i, c in enumerate(node.cases):
                    yield prefix + '|c%s\\' % i
                    for entry in c[1].graph(prefix='|  ' + prefix):
                        yield entry
                    yield prefix + '|<--'
            else:
                yield prefix + '|-' + node.name

    def graph_dot(self, end=''):
        """
        Generate a compatible dot graph for this channel.
        """
        after = []
        cases = None

        yield '#---'

        previous = self.name

        if end == '':
            end = self.name

        for node in self._nodes:
            if isinstance(node, SubChannel):
                yield '"%s"->"%s";' % (previous, node.name)
                after.append((None, node))

            elif isinstance(node, ConditionSubChannel):
                yield '"%s"->"%s" [style=dotted];' % (previous, node.name)
                after.append((end, node))

            elif isinstance(node, Case):
                cases = [c[1] for c in node.cases]

            else:
                if cases:
                    for c in cases:
                        yield '"%s"->"%s" [style=dotted];' % (previous, c.name)
                        after.append((node.name, c))
                    cases = None

                else:
                    yield '"%s"->"%s";' % (previous, node.name)

                previous = node.name

        if end:
            yield '"%s"->"%s";' % (previous, end)

        for end, sub in after:
            for entry in sub.graph_dot(end=end):
                yield entry

    def __str__(self):
        return "<chan: %s>" % self.name

    def __repr__(self):
        typ = type(self)
        return "<%s.%s:(\"%s\")>" % (typ.__module__, typ.__name__, self.name) class PypemanParamError(PypemanError):
    """ custom error """ class TearDownProjectTestCase(TestCase):
    """ unittest for tests creating a project

    this testcase ensures, that at the end of the test the project is
    removed
    """

    def tearDown(self):
        clear_pypeman_project() class ExceptNode(TestNode):
    # This node raises an exception
    def process(self, msg):
        super().process(msg)
        raise TestException() def generate_msg(timestamp=None, message_content=default_message_content,
                 message_meta=None, with_context=False):
    """ generates a default message """
    m = message.Message()
    if timestamp:
        if isinstance(timestamp, tuple):
            m.timestamp = datetime.datetime(*timestamp)
        else:  # assume it's a datetime object
            m.timestamp = timestamp
    else:  # just use current time
        m.timestamp = datetime.datetime.utcnow()

    m.payload = message_content

    if message_meta is None:
        m.meta = dict(default_message_meta)
    else:
        m.meta = message_meta

    if with_context:
        # Add context message
        mctx = generate_msg(message_content={'question': 'known'}, message_meta={'answer': 43})
        m.add_context('test', mctx)

    return m class TestException(Exception):
    """ custom Exception """ class TestNode(nodes.BaseNode):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Used to test if node is processed during test

    def process(self, msg):
        print("Process %s" % self.name)
        return msg
import asyncio
from socket import SOL_SOCKET
from unittest import mock
from pypeman import channels, endpoints
from pypeman import nodes
from pypeman import events
from pypeman.channels import BaseChannel
from pypeman.errors import PypemanParamError
from pypeman.test import TearDownProjectTestCase as TestCase
from pypeman.tests.common import ExceptNode
from pypeman.tests.common import generate_msg
from pypeman.tests.common import TestException
from pypeman.tests.common import TestNode

        msg = generate_msg()

        chan.add(n1)
        sub = chan.fork(name="Hello")
        sub.add(n2, n3)

        # Launch channel processing
        self.start_channels()

        self.loop.run_until_complete(chan.handle(msg))

        self.assertEqual(n1.processed, 1, "Sub Channel not working")

        with self.assertRaises(TestException):
            self.clean_loop()

        self.assertEqual(n2.processed, 1, "Sub Channel not working")

    def test_cond_channel(self):
        """ Whether Conditionnal channel is working """

        chan = BaseChannel(name="test_channel5", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="end_main")
        not_processed = TestNode(name="cond_notproc")
        processed = TestNode(name="cond_proc")

        msg = generate_msg()

        chan.add(n1)

        # Nodes in this channel should not be processed
        cond1 = chan.when(lambda x: False, name="Toto")
        # Nodes in this channel should be processed
        cond2 = chan.when(True, name="condchannel")

        chan.add(n2)

        cond1.add(not_processed)
        cond2.add(processed)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

        self.assertFalse(not_processed.processed, "Cond Channel when condition == False not working")
        self.assertTrue(processed.processed, "Cond Channel when condition == True not working")
        self.assertFalse(n2.processed, "Cond Channel don't became the main path")
        self.assertEqual(cond2.name, "test_channel5.condchannel", "Condchannel name is incorrect")

    def test_case_channel(self):
        """ Whether Conditionnal channel is working """

        chan = BaseChannel(name="test_channel6", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="end_main")
        not_processed = TestNode(name="cond_notproc")
        processed = TestNode(name="cond_proc")
        not_processed2 = TestNode(name="cond_proc2")

        msg = generate_msg()

        chan.add(n1)

        cond1, cond2, cond3 = chan.case(lambda x: False, True, True, names=['first', 'second', 'third'])

        chan.add(n2)

        cond1.add(not_processed)
        cond2.add(processed)
        cond3.add(not_processed2)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

        self.assertFalse(not_processed.processed, "Case Channel when condition == False not working")
        self.assertFalse(not_processed2.processed, "Case Channel when condition == False not working")
        self.assertTrue(processed.processed, "Case Channel when condition == True not working")
        self.assertTrue(n2.processed, "Cond Channel don't became the main path")
        self.assertEqual(cond1.name, "test_channel6.first", "Casechannel name is incorrect")
        self.assertEqual(cond2.name, "test_channel6.second", "Casechannel name is incorrect")
        self.assertEqual(cond3.name, "test_channel6.third", "Casechannel name is incorrect")

    def test_channel_subchannel(self):
        """ Whether BaseChannel subchannel works """
        chan = BaseChannel(name="test_channel6.5", loop=self.loop)

        chan_fork = chan.fork()

        chan_when = chan_fork.when(lambda: True)

        chan_case1, chan_case2 = chan_when.case(lambda: True, lambda: False)

        print(chan.subchannels())

        self.assertEqual(
            len(chan.subchannels()[0]['subchannels'][0]['subchannels']),
            2, "Subchannel graph not working")

    def test_channel_result(self):
        """ Whether BaseChannel handling return a good result """

        chan = BaseChannel(name="test_channel7", loop=self.loop)
        msg = generate_msg()

        chan.add(nodes.JsonToPython(), nodes.PythonToJson())

        # Launch channel processing
        self.start_channels()
        result = self.loop.run_until_complete(chan.handle(msg))

        self.assertEqual(result.payload, msg.payload, "Channel handle not working")

    def test_channel_with_generator(self):
        """ Whether BaseChannel with generator is working """

        chan = BaseChannel(name="test_channel7.3", loop=self.loop)
        chan2 = BaseChannel(name="test_channel7.31", loop=self.loop)

        msg = generate_msg()