Disallow combining of DaskStream and Stream objects

streamz/core.py

@@ -105,11 +105,18 @@ class Stream(object):
 
     def __init__(self, upstream=None, upstreams=None, stream_name=None,
                  loop=None, asynchronous=None, ensure_io_loop=False):
+        from .dask import DaskStream
+
         self.downstreams = OrderedWeakrefSet()
         if upstreams is not None:
             self.upstreams = list(upstreams)
         else:
             self.upstreams = [upstream]
+        if upstreams:
+            if (all(isinstance(u, DaskStream) for u in upstreams) !=
+                    any(isinstance(u, DaskStream) for u in upstreams)):
+                raise ValueError('Must combine only DaskStreams '
+                                 'or no DaskStreams')
 
         self._set_asynchronous(asynchronous)
         self._set_loop(loop)

streamz/tests/test_dask.py

@@ -178,3 +178,12 @@ def add(x, y, z=0):
         yield source.emit((i, i))
 
     assert L == [10, 12, 14, 16, 18]
+
+
+def test_dask_not_dask():
+    with cluster() as (s, [a, b]):
+        with Client(s['address']):
+            s = Stream()
+            s2 = s.scatter()
+            with pytest.raises(ValueError):
+                s.zip(s2)