Uniformisation of API for graph and node contexts (recv -> write), removing publication of LoopbackBag as this is not stable with current BEGIN/END implementation.

bonobo/execution/graph.py

@@ -43,7 +43,7 @@ class GraphExecutionContext:
     def __iter__(self):
         yield from self.nodes
 
-    def recv(self, *messages):
+    def write(self, *messages):
         """Push a list of messages in the inputs of this graph's inputs, matching the output of special node "BEGIN" in
         our graph."""
 
@@ -56,12 +56,12 @@ class GraphExecutionContext:
         for node in self.nodes:
             node.start()
 
-    def loop(self):
-        # todo use strategy
-        for node in self.nodes:
-            node.loop()
-
     def stop(self):
         # todo use strategy
         for node in self.nodes:
             node.stop()
+
+    def loop(self):
+        # todo use strategy
+        for node in self.nodes:
+            node.loop()

bonobo/execution/node.py

@@ -2,7 +2,7 @@ import traceback
 from queue import Empty
 from time import sleep
 
-from bonobo.constants import INHERIT_INPUT, NOT_MODIFIED, BEGIN, END
+from bonobo.constants import INHERIT_INPUT, NOT_MODIFIED
 from bonobo.errors import InactiveReadableError, UnrecoverableError
 from bonobo.execution.base import LoopingExecutionContext
 from bonobo.structs.bags import Bag

bonobo/strategies/executor.py

@@ -21,7 +21,7 @@ class ExecutorStrategy(Strategy):
 
     def execute(self, graph, *args, plugins=None, services=None, **kwargs):
         context = self.create_graph_execution_context(graph, plugins=plugins, services=services)
-        context.recv(BEGIN, Bag(), END)
+        context.write(BEGIN, Bag(), END)
 
         executor = self.create_executor()
 
@@ -57,7 +57,7 @@ class ExecutorStrategy(Strategy):
             futures.append(executor.submit(_runner))
 
         while context.alive:
-            time.sleep(0.2)
+            time.sleep(0.1)
 
         for plugin_context in context.plugins:
             plugin_context.shutdown()

bonobo/strategies/naive.py

@@ -6,7 +6,7 @@ from bonobo.structs.bags import Bag
 class NaiveStrategy(Strategy):
     def execute(self, graph, *args, plugins=None, **kwargs):
         context = self.create_graph_execution_context(graph, plugins=plugins)
-        context.recv(BEGIN, Bag(), END)
+        context.write(BEGIN, Bag(), END)
 
         # TODO: how to run plugins in "naive" mode ?
         context.start()

bonobo/structs/__init__.py

@@ -1,4 +1,4 @@
-from bonobo.structs.bags import Bag, ErrorBag, LoopbackBag
+from bonobo.structs.bags import Bag, ErrorBag
 from bonobo.structs.graphs import Graph
 from bonobo.structs.tokens import Token
 
@@ -6,6 +6,5 @@ __all__ = [
     'Bag',
     'ErrorBag',
     'Graph',
-    'LoopbackBag',
     'Token',
 ]

tests/test_commands.py

@@ -103,8 +103,8 @@ def test_version(runner, capsys):
 def test_run_with_env(runner, capsys):
     runner(
         'run', '--quiet',
-        get_examples_path('env_vars/get_passed_env.py'), '--env', 'ENV_TEST_NUMBER=123',
-        '--env', 'ENV_TEST_USER=cwandrews', '--env', "ENV_TEST_STRING='my_test_string'"
+        get_examples_path('env_vars/get_passed_env.py'), '--env', 'ENV_TEST_NUMBER=123', '--env',
+        'ENV_TEST_USER=cwandrews', '--env', "ENV_TEST_STRING='my_test_string'"
     )
     out, err = capsys.readouterr()
     out = out.split('\n')

tests/test_execution.py

@@ -62,7 +62,7 @@ def test_simple_execution_context():
     assert not ctx.started
     assert not ctx.stopped
 
-    ctx.recv(BEGIN, Bag(), END)
+    ctx.write(BEGIN, Bag(), END)
 
     assert not ctx.alive
     assert not ctx.started