From cf3d5db28ea3a47ee65ef6e2af8545721ec0af5e Mon Sep 17 00:00:00 2001
From: Thilo Behnke <thilo.behnke@gmx.net>
Date: Sun, 10 Jul 2022 14:59:56 +0200
Subject: [PATCH] problem: rskafka only fetches one record per topic

---
 .../src/components/InstrumentedTickWrapper.svelte     | 11 ++++++++++-
 client/svelte-client/src/store/model/event.ts         |  2 +-
 server/src/kafka.rs                                   |  2 +-
 3 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/client/svelte-client/src/components/InstrumentedTickWrapper.svelte b/client/svelte-client/src/components/InstrumentedTickWrapper.svelte
index c244c9d..0a7d6f1 100644
--- a/client/svelte-client/src/components/InstrumentedTickWrapper.svelte
+++ b/client/svelte-client/src/components/InstrumentedTickWrapper.svelte
@@ -10,10 +10,19 @@
     let state: GameState;
     $: state = $gameStateEvents;
 
+    let lastTick;
+
     $: if (networkTickEvents && $networkTickEvents.hasNext) {
         const tick = networkTickEvents.next();
         if (tick != null) {
-            gameField.update(tick.objects, state);
+            console.warn(`Received tick: ${tick.tick}`)
+            if (lastTick && lastTick.tick >= tick.tick) {
+                console.error(`???? DUPLICATED TICK: ${JSON.stringify(tick)} (vs ${lastTick.tick}) ????`)
+            } else {
+                console.warn(`!!!! Valid tick: ${JSON.stringify(tick)} !!!!`)
+                gameField.update(tick.objects, state);
+                lastTick = tick;
+            }
         }
     }
 
diff --git a/client/svelte-client/src/store/model/event.ts b/client/svelte-client/src/store/model/event.ts
index 0c712a6..839b0ff 100644
--- a/client/svelte-client/src/store/model/event.ts
+++ b/client/svelte-client/src/store/model/event.ts
@@ -25,7 +25,7 @@ export type InputEventPayload = {
 export type TickEventPayload = {
     session_id: string,
     objects: GameObject[],
-    ts: number
+    tick: number
 }
 
 export type StatusEventPayload = {
diff --git a/server/src/kafka.rs b/server/src/kafka.rs
index 17f81d6..7d6f7c8 100644
--- a/server/src/kafka.rs
+++ b/server/src/kafka.rs
@@ -150,7 +150,7 @@ impl KafkaEventReaderImpl {
     async fn consume(&mut self) -> Result<Vec<EventWrapper>, String> {
         debug!("kafka consumer called to consume messages for {:?} / {:?}", self.topic, self.partition);
         // TODO: Only 1 message?
-        let next_res = tokio::time::timeout(Duration::from_millis(1), self.consumer.next()).await;
+        let next_res = tokio::time::timeout(Duration::from_millis(3), self.consumer.next()).await;
         if let Err(e) = next_res {
             info!("No record received in time after {}, timeout for {} / {}.", e, self.topic, self.partition);
             return Ok(vec![]);