morefix

Author: Martin Durant

streamz/core.py

@@ -177,6 +177,7 @@ class Stream(object):
 
     def __init__(self, upstream=None, upstreams=None, stream_name=None,
                  loop=None, asynchronous=None, ensure_io_loop=False):
+        self.name = stream_name
         self.downstreams = OrderedWeakrefSet()
         if upstreams is not None:
             self.upstreams = list(upstreams)
@@ -196,8 +197,6 @@ def __init__(self, upstream=None, upstreams=None, stream_name=None,
             if upstream:
                 upstream.downstreams.add(self)
 
-        self.name = stream_name
-
     def _set_loop(self, loop):
         self.loop = None
         if loop is not None:
@@ -1008,7 +1007,8 @@ def __init__(self, upstream, n, timeout=None, key=None, **kwargs):
         self._buffer = defaultdict(lambda: [])
         self._metadata_buffer = defaultdict(lambda: [])
         self._callbacks = {}
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
 
     def _get_key(self, x):
         if self._key is None:
@@ -1220,7 +1220,8 @@ def __init__(self, upstream, interval, **kwargs):
         self.metadata_buffer = []
         self.last = gen.moment
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
 
         self.loop.add_callback(self.cb)
 
@@ -1322,7 +1323,8 @@ def __init__(
         self._buffer = {}
         self._metadata_buffer = {}
         self.last = gen.moment
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
         self.loop.add_callback(self.cb)
 
     def _get_key(self, x):
@@ -1369,7 +1371,8 @@ def __init__(self, upstream, interval, **kwargs):
         self.interval = convert_interval(interval)
         self.queue = Queue()
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream,**kwargs)
 
         self.loop.add_callback(self.cb)
 
@@ -1407,7 +1410,8 @@ def __init__(self, upstream, interval, **kwargs):
         self.interval = convert_interval(interval)
         self.next = 0
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
 
     @gen.coroutine
     def update(self, x, who=None, metadata=None):
@@ -1432,7 +1436,8 @@ class buffer(Stream):
     def __init__(self, upstream, n, **kwargs):
         self.queue = Queue(maxsize=n)
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
 
         self.loop.add_callback(self.cb)
 
@@ -1876,7 +1881,8 @@ def __init__(self, upstream, **kwargs):
         self.next = []
         self.next_metadata = None
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
 
         self.loop.add_callback(self.cb)
 
@@ -1932,7 +1938,8 @@ def __init__(self, upstream, topic, producer_config, **kwargs):
         self.topic = topic
         self.producer = ck.Producer(producer_config)
 
-        Stream.__init__(self, upstream, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        Stream.__init__(self, upstream, **kwargs)
         self.stopped = False
         self.polltime = 0.2
         self.loop.add_callback(self.poll)

streamz/dask.py

@@ -38,7 +38,8 @@ class DaskStream(Stream):
     dask.distributed.Client
     """
     def __init__(self, *args, **kwargs):
-        super().__init__(*args, ensure_io_loop=True, **kwargs)
+        kwargs["ensure_io_loop"] = True
+        super().__init__(*args, **kwargs)
 
 
 @DaskStream.register_api()

streamz/tests/test_core.py

@@ -403,7 +403,7 @@ def test_timed_window():
 
 @gen_test()
 def test_timed_window_ref_counts():
-    source = Stream()
+    source = Stream(asynchronous=True)
     _ = source.timed_window(0.01)
 
     ref1 = RefCounter()

streamz/tests/test_dask.py

@@ -71,7 +71,6 @@ def test_partition_then_scatter_async(c, s, a, b):
     assert L == [1, 2, 3]
 
 
-@pytest.mark.slow
 def test_partition_then_scatter_sync(loop):
     # Ensure partition w/ timeout before scatter works correctly for synchronous
     with cluster() as (s, [a, b]):
@@ -163,7 +162,6 @@ def test_accumulate(c, s, a, b):
     assert L[-1][1] == 3
 
 
-@pytest.mark.slow
 def test_sync(loop):  # noqa: F811
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
@@ -180,7 +178,6 @@ def f():
             assert L == list(map(inc, range(10)))
 
 
-@pytest.mark.slow
 def test_sync_2(loop):  # noqa: F811
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop):  # noqa: F841
@@ -194,7 +191,6 @@ def test_sync_2(loop):  # noqa: F811
             assert L == list(map(inc, range(10)))
 
 
-@pytest.mark.slow
 @gen_cluster(client=True, ncores=[('127.0.0.1', 1)] * 2)
 def test_buffer(c, s, a, b):
     source = Stream(asynchronous=True)
@@ -221,7 +217,6 @@ def test_buffer(c, s, a, b):
     assert source.loop == c.loop
 
 
-@pytest.mark.slow
 def test_buffer_sync(loop):  # noqa: F811
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as c:  # noqa: F841
@@ -246,7 +241,6 @@ def test_buffer_sync(loop):  # noqa: F811
 
 
 @pytest.mark.xfail(reason='')
-@pytest.mark.slow
 def test_stream_shares_client_loop(loop):  # noqa: F811
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841