Repository: mhcomm/pypeman

Issue:
Dropnode breaks message generator
When you use a node that return a message generator, appending a dropnode raise an exception that break the loop to consume all message generator. We should wait for the last message to reraise the exception.

Discussion:

Patch:
diff --git a/pypeman/channels.py b/pypeman/channels.py
index e02ca05..e448dec 100644
--- a/pypeman/channels.py
+++ b/pypeman/channels.py
@@ -312,8 +312,13 @@ class BaseChannel:
 
         if self.next_node:
             if isinstance(result, types.GeneratorType):
-                for res in result:
-                    result = await self.next_node.handle(res)
+                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)
diff --git a/pypeman/commands.py b/pypeman/commands.py
index 3225205..bef7cb2 100755
--- a/pypeman/commands.py
+++ b/pypeman/commands.py
@@ -117,7 +117,7 @@ def mk_daemon(mainfunc=lambda: None, pidfile="pypeman.pid"):
     # - daemonocle
     # - py daemoniker
     # Alternatively if we don't want other module dependencies we might just copy
-    # the DaemonLite files into your source repository 
+    # the DaemonLite files into your source repository
     class DaemonizedApp(DaemonLite):
         def run(self):
             mainfunc()
@@ -127,20 +127,17 @@ def mk_daemon(mainfunc=lambda: None, pidfile="pypeman.pid"):
 
 @begin.subcommand
 def start(reload: 'Make server autoreload (Dev only)'=False,
-        debug_asyncio: 'Enable asyncio debug'=False,
-        cli : "enables an IPython CLI for debugging (not operational)"=False,
-        profile : "enables profiling / run stats (not operational)"=False,
-        # TODO: can be True if DaemonLite is a hard requirement
-        daemon : "if true pypeman will be started as daemon "=bool(DaemonLite),
-
+          debug_asyncio: 'Enable asyncio debug'=False,
+          cli : "enables an IPython CLI for debugging (not operational)"=False,
+          profile : "enables profiling / run stats (not operational)"=False,
+          daemon : "if true pypeman will be started as daemon "=True,
         ):
     """ Start pypeman as daemon (or foreground process) """
-    main_func = partial(main, debug_asyncio=debug_asyncio, cli=cli, 
+    main_func = partial(main, debug_asyncio=debug_asyncio, cli=cli,
             profile=profile)
     start_func = partial(reloader_opt, main_func, reload, 2)
 
     if reload:
-        print("RELOAD")
         start_func()
     else:
         if daemon:
diff --git a/pypeman/contrib/http.py b/pypeman/contrib/http.py
index 3658dc2..f55bcea 100644
--- a/pypeman/contrib/http.py
+++ b/pypeman/contrib/http.py
@@ -17,21 +17,32 @@ class HTTPEndpoint(endpoints.BaseEndpoint):
     def __init__(self, adress='127.0.0.1', port='8080', loop=None, http_args=None):
         super().__init__()
         self.http_args = http_args or {}
+        self.ssl_context = http_args.pop('ssl_context', None)
+
         self._app = None
         self.address = adress
         self.port = port
         self.loop = loop or asyncio.get_event_loop()
 
-    def add_route(self,*args, **kwargs):
-        if not self._app:
-            self._app = web.Application(loop=self.loop, **self.http_args)
-        # TODO route should be added later
+    def add_route(self, *args, **kwargs):
+        if self._app is None:
+            self._app = web.Application(**self.http_args)
+
         self._app.router.add_route(*args, **kwargs)
 
     async def start(self):
         if self._app is not None:
-            srv = await self.loop.create_server(self._app.make_handler(), self.address, self.port)
-            print("Server started at http://{}:{}".format(self.address, self.port))
+            srv = await self.loop.create_server(
+                protocol_factory=self._app.make_handler(),
+                host=self.address,
+                port=self.port,
+                ssl=self.ssl_context
+            )
+            print("Server started at http{}://{}:{}".format(
+                's' if self.ssl_context else '',
+                self.address,
+                self.port
+            ))
             return srv
         else:
             print("No HTTP route.")
diff --git a/pypeman/helpers/reloader.py b/pypeman/helpers/reloader.py
index f675d1e..cadbd4e 100644
--- a/pypeman/helpers/reloader.py
+++ b/pypeman/helpers/reloader.py
@@ -45,7 +45,7 @@ def reloader_opt(to_call, reloader, interval):
         lockfile = None
         try:
             fd, lockfile = tempfile.mkstemp(prefix='process.', suffix='.lock')
-            os.close(fd)  # We only need this file to exist. We never write to it
+            os.close(fd)  # We only need this file to exist. We never write to it.
             while os.path.exists(lockfile):
                 args = [sys.executable] + sys.argv
                 environ = os.environ.copy()
diff --git a/pypeman/nodes.py b/pypeman/nodes.py
index 67c7452..d13d4a8 100644
--- a/pypeman/nodes.py
+++ b/pypeman/nodes.py
@@ -98,7 +98,7 @@ class BaseNode:
         if log_output:
             # Enable logging
             self._handle_without_log = self.handle
-            self.handle = self._log_handle
+            setattr(self, 'handle', self._log_handle)
 
     async def handle(self, msg):
         """ Handle message is called by channel to launch process method on it.
@@ -129,8 +129,13 @@ class BaseNode:
 
         if self.next_node:
             if isinstance(result, types.GeneratorType):
-                for res in result:
-                    result = await self.next_node.handle(res)
+                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:
                 if self.store_output_as:
@@ -225,7 +230,7 @@ class BaseNode:
 
         if not hasattr(self, '_handle'):
             self._handle = self.handle
-            self.handle = self._test_handle
+            setattr(self, 'handle', self._test_handle)
 
         if hasattr(self, '_orig_process'):
             self.process = self._orig_process


Test patch:
diff --git a/pypeman/tests/test_channel.py b/pypeman/tests/test_channel.py
index 6020e70..ba0b42a 100644
--- a/pypeman/tests/test_channel.py
+++ b/pypeman/tests/test_channel.py
@@ -230,6 +230,39 @@ class ChannelsTests(unittest.TestCase):
 
         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()
+        msg2 = msg.copy()
+
+        class TestIter(nodes.BaseNode):
+            def process(self, msg):
+                def iter():
+                    for i in range(3):
+                        yield msg
+                return iter()
+
+        final_node = nodes.Log()
+        mid_node = nodes.Log()
+
+        chan.add(TestIter(name="testiterr"), nodes.Log(), TestIter(name="testiterr"), final_node)
+        chan2.add(TestIter(name="testiterr"), mid_node, nodes.Drop())
+
+        # Launch channel processing
+        self.start_channels()
+        result = self.loop.run_until_complete(chan.handle(msg))
+
+        self.assertEqual(result.payload, msg.payload, "Generator node not working")
+        self.assertEqual(final_node.processed, 9, "Generator node not working")
+
+        result = self.loop.run_until_complete(chan2.handle(msg2))
+
+        self.assertEqual(mid_node.processed, 3, "Generator node not working with drop_node")
+
+
     def test_channel_events(self):
         """ Whether BaseChannel handling return a good result """