nebkor
/
chatty-catties
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

actually implement list_committed_offsets

This commit is contained in:
Joe Ardent 2024-06-07 13:12:37 -07:00
parent 5222d55a61
commit a492ca56a4
1 changed files with 30 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
gg-kafka/src/main.rs
View File

@ -17,11 +17,16 @@ fn on_init(runner: &Runner) {
} }
#[derive(Debug, Default)] #[derive(Debug, Default)]
struct Roach { struct OffsetState {
// these first four get written to as messages come in last_page: usize,
logs: Logs,
last_pages: HashMap<String, u64>,
current_chapter: usize, current_chapter: usize,
}
#[derive(Debug, Default)]
struct Roach {
// these first three get written to as messages come in
logs: Logs,
state: HashMap<String, OffsetState>,
committed_offsets: HashMap<String, u64>, committed_offsets: HashMap<String, u64>,
// these are initialized on init and then just read // these are initialized on init and then just read
@ -39,17 +44,17 @@ impl Roach {
} }
fn next_offset(&mut self, key: &str) -> u64 { fn next_offset(&mut self, key: &str) -> u64 {
let last_page = self.last_pages.entry(key.to_string()).or_default(); let state = self.state.entry(key.to_string()).or_default();
if *last_page + 1 < self.pages as u64 { if state.last_page + 1 < self.pages {
*last_page += 1; state.last_page += 1;
} else { } else {
*last_page = 0; state.last_page = 0;
self.current_chapter += 1; state.current_chapter += 1;
} }
let cur_chap = self.current_chapter; let cur_chap = state.current_chapter;
let offset_base = self.offset_base; let offset_base = self.offset_base;
let start = (offset_base * self.pages) + (cur_chap * (self.pages * self.sub_chapters)); let start = (offset_base * self.pages) + (cur_chap * (self.pages * self.sub_chapters));
start as u64 + *last_page (start + state.last_page) as u64
} }
fn init_log(&mut self, runner: &Runner) { fn init_log(&mut self, runner: &Runner) {
@ -113,6 +118,18 @@ impl Node for Roach {
runner.reply(&msg, body); runner.reply(&msg, body);
} }
"commit_offsets" => { "commit_offsets" => {
let offsets = msg
.body
.payload
.get("offsets")
.unwrap()
.as_object()
.unwrap();
for (k, v) in offsets.iter() {
self.committed_offsets
.entry(k.to_string())
.or_insert(v.as_u64().unwrap());
}
runner.reply(&msg, Body::from_type("commit_offsets_ok")); runner.reply(&msg, Body::from_type("commit_offsets_ok"));
} }
"list_committed_offsets" => { "list_committed_offsets" => {
@ -123,6 +140,8 @@ impl Node for Roach {
.collect::<Payload>(); .collect::<Payload>();
let mut payload = Payload::new(); let mut payload = Payload::new();
payload.insert("offsets".to_string(), offsets.into()); payload.insert("offsets".to_string(), offsets.into());
let body = Body::from_type("list_committed_offsets_ok").with_payload(payload);
runner.reply(&msg, body);
} }
_ => eprintln!("unknown Message type: {msg:?}"), _ => eprintln!("unknown Message type: {msg:?}"),
} }