emit empty tuple on subscription of CombineLatest without publishers

fix bug for allow_stateless when all publishers are stateful

broqer/op/combine_latest.py

@@ -92,6 +92,12 @@ def subscribe(self, subscriber: Subscriber,
 
         disposable = MultiOperator.subscribe(self, subscriber, prepend)
 
+        # if there are no source publishers emit an empty tuple on subscription
+        if not self._publishers:
+            self.notify(())
+
+        # check if ._statless is already definied (will be done on first
+        # subscription)
         if self._stateless is not None:
             return disposable
 
@@ -165,7 +171,7 @@ def emit(self, value: Any, who: Publisher) -> asyncio.Future:
             state = tuple(self._partial_state)
 
         # remove stateless publisher emits from ._partial_state
-        if self._stateless[index]:
+        if self._stateless and self._stateless[index]:
             self._partial_state[index] = NONE
 
         # if result of _map() was NONE don't emit
@@ -176,7 +182,7 @@ def emit(self, value: Any, who: Publisher) -> asyncio.Future:
         is_new_state = (state == self._state)
 
         # store ._state only when all publishers are stateful
-        if not any(self._stateless):
+        if self._stateless and not any(self._stateless):
             self._state = state
 
         # check if state has changed or stateless publisher has emitted

test/test_op_combine_latest.py

@@ -218,4 +218,26 @@ def reverse(s1, s2, s3, s4):
     source4.notify(4)
     with pytest.raises(ValueError):
         dut.get()
-    assert collector.result_vector == ((NONE, 3, NONE, 1), (4, 3, NONE, 1),)
+    assert collector.result_vector == ((NONE, 3, NONE, 1), (4, 3, NONE, 1),)
+
+def test_allow_stateless_with_stateful_publishers():
+    source1 = StatefulPublisher(0)
+    source2 = StatefulPublisher(0)
+
+    dut = CombineLatest(source1, source2, allow_stateless=True)
+
+    collector = Collector()
+    dut.subscribe(collector)
+
+    source1.notify(1)
+    source2.notify(1)
+
+    assert collector.result_vector == ((0,0), (1,0), (1,1))
+
+def test_no_publishers():
+    dut = CombineLatest()
+
+    collector = Collector()
+    dut.subscribe(collector)
+
+    assert collector.result_vector == ((),)