Introduce small delay to avoid race between publisher and subscriber

Signed-off-by: Tim Paine <[email protected]>

Author: timkpaine

csp/tests/adapters/test_kafka.py

@@ -131,6 +131,7 @@ def graph(symbols: list, count: int):
                 csp.timer(timedelta(seconds=0.2), True),
                 csp.delay(csp.timer(timedelta(seconds=0.2), False), timedelta(seconds=0.1)),
             )
+
             i = csp.count(csp.timer(timedelta(seconds=0.15)))
             d = csp.count(csp.timer(timedelta(seconds=0.2))) / 2.0
             s = csp.sample(csp.timer(timedelta(seconds=0.4)), csp.const("STRING"))
@@ -157,18 +158,13 @@ def graph(symbols: list, count: int):
                 )
                 csp.add_graph_output(f"pall_{symbol}", pub_data)
 
-                # csp.print('status', kafkaadapter.status())
-
                 sub_data = kafkaadapter.subscribe(
                     ts_type=SubData,
                     msg_mapper=msg_mapper,
                     topic=topic,
                     key=symbol,
                     push_mode=csp.PushMode.NON_COLLAPSING,
                 )
-
-                sub_data = csp.firstN(sub_data, count)
-
                 csp.add_graph_output(f"sall_{symbol}", sub_data)
 
                 done_flag = csp.count(sub_data) == count
@@ -190,8 +186,12 @@ def graph(symbols: list, count: int):
             pub = results[f"pall_{symbol}"]
             sub = results[f"sall_{symbol}"]
 
+            # limit by the last `count`
+            sub = sub[-1*count:]
+            pub = pub[-1*count:]
+
             assert len(sub) == count
-            assert [v[1] for v in sub] == [v[1] for v in pub[:count]]
+            assert [v[1] for v in sub] == [v[1] for v in pub[-1*count:]]
 
     @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
     def test_start_offsets(self, kafkaadapter, kafkabroker):