Added some helper component composition functions.

3 files changed, 37 insertions, 6 deletions

diff --git a/README.md b/README.md
index c7b4e5b..4e9a6bf 100644
--- a/README.md
+++ b/README.md
@@ -154,3 +154,17 @@ Construct a wire based on a *conversion* dictionary. Assuming that dictionaries
     helpers.wire_components(component_one, component_two, wire)
 
 Returns a pipeline component that is the composition of two components with a wire between them.
+
+### Component Composition Functions
+
+#### Constructing a Composed Component
+
+    helpers.cons_composed_component(first_component, second_component)
+
+Returns a components that is the composition of the `first_component` and the `second_component`.
+
+#### Constructing a Parallel Component
+
+    helpers.cons_parallel_component(top_component, bottom_component)
+
+Returns a component that will execute the two provided components in parallel. The input to the constructed component is a pair, whose first value is applied to the `top_component` and the second value is applied to the `bottom_component`. The constructed component`s output shall be a pair, whose first value is the output of the top component, and the second value is the output of the bottom component.
diff --git a/src/pypeline/helpers/helpers.py b/src/pypeline/helpers/helpers.py
index 848e804..d811d44 100644
--- a/src/pypeline/helpers/helpers.py
+++ b/src/pypeline/helpers/helpers.py
@@ -170,8 +170,8 @@ def cons_unsplit_wire(unsplit_func):
     return unsplit(return_, unsplit_func)
 
 
-def wire_components(component_one, component_two, wire):
-    """Wire two components together."""
+def cons_wired_components(component_one, component_two, wire):
+    """Wire two components together and return a component that is the composition of these components."""
     return component_one >> wire >> component_two
 
 
@@ -180,6 +180,16 @@ def cons_pipeline(input_wire, component, output_wire):
     return input_wire >> component >> output_wire
 
 
+def cons_composed_component(first_component, second_component):
+    """Compose two components and return a component that represents the composed computation."""
+    return first_component >> second_component
+
+
+def cons_parallel_component(top_component, bottom_component):
+    """Construct a component that will compute the provided components in parallel. The returned component takes a pair as input, see cons_split_wire(), and the component shall return a pair."""
+    return top_component ** bottom_component
+
+
 def __kleisli_wrapper(f):
     def wrapper(pipeline, input, state):
         """Run, evaluate, or execute a pipeline."""
diff --git a/src/pypeline/helpers/tests/helper_pipeline_tests.py b/src/pypeline/helpers/tests/helper_pipeline_tests.py
index d34f878..e660108 100644
--- a/src/pypeline/helpers/tests/helper_pipeline_tests.py
+++ b/src/pypeline/helpers/tests/helper_pipeline_tests.py
@@ -34,7 +34,9 @@ from pypeline.helpers.helpers import cons_subprocess_component, \
      cons_split_wire, \
      cons_unsplit_wire, \
      cons_pipeline, \
-     wire_components, \
+     cons_wired_components, \
+     cons_composed_component, \
+     cons_parallel_component, \
      run_pipeline
 
 
@@ -103,7 +105,10 @@ class PypelineHelperFunctionUnitTest(unittest.TestCase):
                     output_wire_func = lambda a, s: str(a['output'])
                     output_wire = cons_wire(output_wire_func)
 
-                    pipeline = input_wire >> wire_components(comp_one, comp_two, wire) >> to_upper_wire >> comp_three
+                    pipeline = cons_pipeline(input_wire,
+                                             cons_wired_components(comp_one, comp_two, wire),
+                                             to_upper_wire)
+                    pipeline = cons_composed_component(pipeline, comp_three)
 
                     value = "hello world"
                     target = (upper_func(value, None), [rev_msg_one, rev_msg_two, upper_msg])
@@ -144,13 +149,15 @@ class PypelineHelperFunctionUnitTest(unittest.TestCase):
                     output_func,
                     state_mutator = lambda s: s.append(rev_msg_two) or s)
 
-               parallel_reverse_comp = comp_one ** comp_two
+               parallel_reverse_comp = cons_parallel_component(comp_one, comp_two)
                split_wire = cons_split_wire()
                unsplit_func = lambda a, b: {'subprocess_output' : a['output'],
                                             'function_output': b['output']}
                unsplit_wire = cons_unsplit_wire(unsplit_func)
                input_wire = cons_wire(lambda a, s: {'input': a})
-               pipeline = input_wire >> split_wire >> parallel_reverse_comp >> unsplit_wire
+               pipeline = cons_pipeline(input_wire,
+                                        cons_composed_component(split_wire, parallel_reverse_comp),
+                                        unsplit_wire)
 
                value = "hello world"
                result = run_pipeline(pipeline, "hello world", list())