Introduce small delay to avoid race between publisher and subscriber

Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>

Author: timkpaine

csp/tests/adapters/test_kafka.py

@@ -127,13 +127,18 @@ def curtime(x: ts[object]) -> ts[datetime]:
                 return csp.now()
 
         def graph(symbols: list, count: int):
-            b = csp.merge(
-                csp.timer(timedelta(seconds=0.2), True),
-                csp.delay(csp.timer(timedelta(seconds=0.2), False), timedelta(seconds=0.1)),
+            delay = timedelta(seconds=1)
+            b = csp.delay(
+                csp.merge(
+                    csp.timer(timedelta(seconds=0.2), True),
+                    csp.delay(csp.timer(timedelta(seconds=0.2), False), timedelta(seconds=0.1)),
+                ),
+                delay=delay,
             )
-            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"))
+
+            i = csp.delay(csp.count(csp.timer(timedelta(seconds=0.15))), delay=delay)
+            d = csp.delay(csp.count(csp.timer(timedelta(seconds=0.2))) / 2.0, delay=delay)
+            s = csp.delay(csp.sample(csp.timer(timedelta(seconds=0.4)), csp.const("STRING")), delay=delay)
             dt = curtime(b)
             struct = MyData.collectts(b=b, i=i, d=d, s=s, dt=dt)
 
@@ -157,18 +162,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 +190,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):