Add Port object

Author: samuell

examples/string_processing_pipeline.py

@@ -1,4 +1,3 @@
-import asyncio
 import flowbase
 
 
@@ -37,17 +36,17 @@ def main():
 
 
 class HiSayer:
-    out_lines = asyncio.Queue()
+    out_lines = flowbase.Port()
 
     async def run(self):
         for i in range(20):
             await self.out_lines.put(f"Hi hi for the {i+1}:th time...")
 
 
 class StringSplitter:
-    in_lines = asyncio.Queue()
-    out_leftpart = asyncio.Queue()
-    out_rightpart = asyncio.Queue()
+    in_lines = flowbase.Port()
+    out_leftpart = flowbase.Port()
+    out_rightpart = flowbase.Port()
 
     async def run(self):
         while not self.in_lines.empty():
@@ -57,8 +56,8 @@ async def run(self):
 
 
 class LowerCaser:
-    in_lines = asyncio.Queue()
-    out_lines = asyncio.Queue()
+    in_lines = flowbase.Port()
+    out_lines = flowbase.Port()
 
     async def run(self):
         while not self.in_lines.empty():
@@ -67,8 +66,8 @@ async def run(self):
 
 
 class UpperCaser:
-    in_lines = asyncio.Queue()
-    out_lines = asyncio.Queue()
+    in_lines = flowbase.Port()
+    out_lines = flowbase.Port()
 
     async def run(self):
         while not self.in_lines.empty():
@@ -77,9 +76,9 @@ async def run(self):
 
 
 class StringJoiner:
-    in_leftpart = asyncio.Queue()
-    in_rightpart = asyncio.Queue()
-    out_lines = asyncio.Queue()
+    in_leftpart = flowbase.Port()
+    in_rightpart = flowbase.Port()
+    out_lines = flowbase.Port()
 
     async def run(self):
         while not self.in_leftpart.empty() or not self.in_rightpart.empty():
@@ -89,7 +88,7 @@ async def run(self):
 
 
 class Printer:
-    in_lines = asyncio.Queue()
+    in_lines = flowbase.Port()
 
     async def run(self):
         while not self.in_lines.empty():

flowbase/__init__.py

@@ -1 +1 @@
-from flowbase.flowbase import Process, Network
+from flowbase.flowbase import Process, Network, Port

flowbase/flowbase.py

@@ -27,3 +27,7 @@ def add_process(self, name: str, process: Process):
 
     def run(self):
         self._loop.run_until_complete(self._driver_process.run())
+
+
+class Port(asyncio.Queue):
+    pass