From 4c89bf7a5c0b1884cd473b20321501a8131ec2bb Mon Sep 17 00:00:00 2001
From: Joe Ardent <code@ardent.nebcorp.com>
Date: Tue, 4 Jun 2024 12:58:38 -0700
Subject: [PATCH] passes global counter

---
 gg-g_counter/src/main.rs | 43 ++++++++++++++++++++++++++++++++--------
 1 file changed, 35 insertions(+), 8 deletions(-)

diff --git a/gg-g_counter/src/main.rs b/gg-g_counter/src/main.rs
index 2d9922b..bcaf4a9 100644
--- a/gg-g_counter/src/main.rs
+++ b/gg-g_counter/src/main.rs
@@ -3,10 +3,10 @@ use nebkor_maelstrom::{kv::Kv, mk_payload, Body, Message, Node, Runner};
 const KEY: &str = "COUNTER";
 
 #[derive(Clone, Default)]
-struct Counter;
+struct Counter(i64);
 
 fn main() {
-    let node = Counter;
+    let node = Counter(0);
 
     let runner = Runner::new(node);
 
@@ -20,6 +20,14 @@ fn main() {
     runner.run(Some(on_init));
 }
 
+fn gossip(neighbors: &[&String], runner: &Runner, val: i64) {
+    let payload = mk_payload(&[("goss", val.into())]);
+    let body = Body::from_type("gossip").with_payload(payload);
+    for node in neighbors {
+        runner.send(node, body.clone());
+    }
+}
+
 impl Node for Counter {
     fn handle(&mut self, runner: &Runner, req: Message) {
         let typ = req.body.typ.as_str();
@@ -28,19 +36,38 @@ impl Node for Counter {
             "add" => {
                 let delta = req.body.payload.get("delta").unwrap().as_i64().unwrap();
                 loop {
-                    let cur = kv.read(runner, KEY).unwrap().unwrap().as_i64().unwrap();
-                    match kv.cas(runner, KEY, cur.into(), (cur + delta).into(), false) {
-                        Err(_) => {}
-                        Ok(_) => break,
+                    let cur = self.0;
+                    let new = cur + delta;
+                    match kv.cas(runner, KEY, cur.into(), new.into(), false) {
+                        Err(_e) => {
+                            let cur = kv.read(runner, KEY).unwrap().unwrap().as_i64().unwrap();
+                            self.0 = cur;
+                        }
+                        Ok(_) => {
+                            let nodes = runner
+                                .nodes()
+                                .iter()
+                                .filter(|e| e != &&runner.node_id())
+                                .collect::<Vec<_>>();
+                            self.0 = new;
+                            gossip(&nodes, runner, new);
+                            break;
+                        }
                     }
                 }
                 runner.reply(&req, Body::from_type("add_ok"));
             }
             "read" => {
-                let val = kv.read(runner, KEY).unwrap().unwrap();
-                let body = Body::from_type("read_ok").with_payload(mk_payload(&[("value", val)]));
+                let val = self.0;
+                let body =
+                    Body::from_type("read_ok").with_payload(mk_payload(&[("value", val.into())]));
                 runner.reply(&req, body);
             }
+
+            "gossip" => {
+                let val = req.body.payload.get("goss").unwrap().as_i64().unwrap();
+                self.0 = val.max(self.0);
+            }
             _ => {
                 eprintln!("unknown type: {req:?}");
             }