Moved the location of the limit

elx/runner.py

@@ -69,7 +69,12 @@ def run(
         streams: Optional[List[str]] = None,
         logger: logging.Logger = None,
     ) -> None:
-        asyncio.run(self.async_run(streams=streams, logger=logger))
+        asyncio.get_event_loop().run_until_complete(
+            self.async_run(
+                streams=streams,
+                logger=logger,
+            )
+        )
 
     async def async_run(
         self,
@@ -95,11 +100,9 @@ def writelines(self, line: str):
         async with self.tap.process(
             state=state,
             streams=streams,
-            limit=self.DEFAULT_BUFFER_SIZE_LIMIT,
         ) as tap_process:
             async with self.target.process(
                 tap_process=tap_process,
-                limit=self.DEFAULT_BUFFER_SIZE_LIMIT,
             ) as target_process:
                 tap_outputs = [target_process.stdin]
                 tap_stdout_future = asyncio.ensure_future(

elx/singer.py

@@ -13,6 +13,7 @@
 from elx.utils import require_install, interpolate_in_config
 
 PYTHON = "python3"
+BUFFER_SIZE_LIMIT = 10485760
 
 
 class Singer:

elx/tap.py

@@ -4,7 +4,7 @@
 from functools import cached_property
 from pathlib import Path
 from typing import Generator, List, Optional
-from elx.singer import Singer, require_install
+from elx.singer import Singer, require_install, BUFFER_SIZE_LIMIT
 from elx.catalog import Stream, Catalog
 from elx.json_temp_file import json_temp_file
 from subprocess import Popen, PIPE
@@ -51,7 +51,6 @@ def catalog(self) -> Catalog:
     @require_install
     async def process(
         self,
-        limit: int,
         state: dict = {},
         streams: Optional[List[str]] = None,
     ) -> Generator[Popen, None, None]:
@@ -78,7 +77,7 @@ async def process(
                         ],
                         stdout=asyncio.subprocess.PIPE,
                         stderr=asyncio.subprocess.PIPE,
-                        limit=limit,
+                        limit=BUFFER_SIZE_LIMIT,
                     )
 
     def invoke(

elx/target.py

@@ -2,7 +2,7 @@
 import contextlib
 from subprocess import PIPE, Popen
 from typing import Generator, Optional
-from elx.singer import Singer, require_install
+from elx.singer import Singer, require_install, BUFFER_SIZE_LIMIT
 from elx.json_temp_file import json_temp_file
 
 
@@ -12,7 +12,6 @@ class Target(Singer):
     async def process(
         self,
         tap_process: Popen,
-        limit: int,
     ) -> Generator[Popen, None, None]:
         """
         Run the tap process.
@@ -36,5 +35,5 @@ async def process(
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
-                limit=limit,
+                limit=BUFFER_SIZE_LIMIT,
             )

tests/test_elx/test_tap.py

@@ -1,6 +1,5 @@
 import asyncio
 import pytest
-from subprocess import Popen
 from elx import Tap
 from elx.catalog import Stream, Catalog
 
@@ -22,6 +21,6 @@ async def test_tap_process(tap: Tap):
     """
     Test that the tap process can be run.
     """
-    async with tap.process(limit=1) as process:
+    async with tap.process() as process:
         # Make sure the tap process is of the right type.
         assert type(process) == asyncio.subprocess.Process