Simplify my fix!

Author: martindurant

streamz/core.py

@@ -1473,18 +1473,23 @@ class zip(Stream):
 
     def __init__(self, *upstreams, **kwargs):
         self.maxsize = kwargs.pop('maxsize', 10)
-        self.condition = Condition()
+        self._condition = None
         self.literals = [(i, val) for i, val in enumerate(upstreams)
                          if not isinstance(val, Stream)]
 
         self.buffers = {upstream: deque()
                         for upstream in upstreams
                         if isinstance(upstream, Stream)}
-
         upstreams2 = [upstream for upstream in upstreams if isinstance(upstream, Stream)]
 
         Stream.__init__(self, upstreams=upstreams2, **kwargs)
 
+    @property
+    def condition(self):
+        if self._condition is None:
+            self._condition = Condition()
+        return self._condition
+
     def _add_upstream(self, upstream):
         # Override method to handle setup of buffer for new stream
         self.buffers[upstream] = deque()
@@ -1881,7 +1886,7 @@ class latest(Stream):
     _graphviz_shape = 'octagon'
 
     def __init__(self, upstream, **kwargs):
-        self.condition = Condition()
+        self._condition = None
         self.next = []
         self.next_metadata = None
 
@@ -1890,6 +1895,12 @@ def __init__(self, upstream, **kwargs):
 
         self.loop.add_callback(self.cb)
 
+    @property
+    def condition(self):
+        if self._condition is None:
+            self._condition = Condition()
+        return self._condition
+
     def update(self, x, who=None, metadata=None):
         if self.next_metadata:
             self._release_refs(self.next_metadata)

streamz/dataframe/tests/test_dataframes.py

@@ -219,7 +219,7 @@ def test_binary_stream_operators(stream):
 
     a.emit(df)
 
-    assert_eq(b[0], expected)
+    wait_for(lambda: b and b[0].equals(expected), 1)
 
 
 def test_index(stream):
@@ -246,7 +246,7 @@ def test_pair_arithmetic(stream):
     a.emit(df.iloc[:5])
     a.emit(df.iloc[5:])
 
-    assert len(L) == 2
+    wait_for(lambda: len(L) == 2, 1)
     assert_eq(pd.concat(L, axis=0), (df.x + df.y) * 2)
 
 
@@ -259,7 +259,7 @@ def test_getitem(stream):
     a.emit(df.iloc[:5])
     a.emit(df.iloc[5:])
 
-    assert len(L) == 2
+    wait_for(lambda: len(L) == 2, 1)
     assert_eq(pd.concat(L, axis=0), df[df.x > 4])
 
 
@@ -298,6 +298,7 @@ def f(x):
     a.emit(df.iloc[7:])
 
     first = df.iloc[:3]
+    wait_for(lambda: len(L) > 2, 1)
     assert assert_eq(L[0], f(first))
     assert assert_eq(L[-1], f(df))
 
@@ -382,7 +383,7 @@ def test_setitem(stream):
     df['a'] = 10
     df[['c', 'd']] = df[['x', 'y']]
 
-    assert_eq(L[-1], df.mean())
+    wait_for(lambda: L and L[-1].equals(df.mean()), 1)
 
 
 def test_setitem_overwrites(stream):