Message queues and event streams

class Queue {
  // A work queue: each message goes to one worker; unacked messages come back; repeated failures go to a DLQ.
  ready: string[] = []; dlq: string[] = []; tries = new Map<string, number>();
  constructor(private maxTries = 3) {}
  send(msg: string) { this.ready.push(msg); }
  receive(): string | undefined { return this.ready.shift(); }
  nack(msg: string) {                      // crash or error: the visibility timeout expires and it comes back
    const n = (this.tries.get(msg) ?? 0) + 1;
    this.tries.set(msg, n);
    (n >= this.maxTries ? this.dlq : this.ready).push(msg);
  }
}

class Log {
  // An event log: messages stay; each consumer group keeps its own offset and can rewind.
  entries: string[] = []; offsets = new Map<string, number>();
  append(msg: string) { this.entries.push(msg); }
  poll(group: string): string[] {
    const start = this.offsets.get(group) ?? 0;
    this.offsets.set(group, this.entries.length);
    return this.entries.slice(start);
  }
  seek(group: string, offset: number) { this.offsets.set(group, offset); }
}

const ORDERS = ["order-1", "order-2", "poison", "order-4"];

const q = new Queue();
for (const o of ORDERS) q.send(o);
const done: string[] = [];
let crashedOnce = false, msg: string | undefined;
while ((msg = q.receive()) !== undefined) {
  if (msg === "poison" || (msg === "order-2" && !crashedOnce)) {
    crashedOnce = crashedOnce || msg === "order-2";
    q.nack(msg);                           // order-2 fails once (a worker crash); poison fails every time
  } else done.push(msg);
}
console.log(`queue: processed ${done.join(", ")} | dead-letter: ${q.dlq.join(", ")} | poison tried ${q.tries.get("poison")} times`);

const log = new Log();
for (const o of ORDERS) log.append(o);
console.log("log: billing reads", log.poll("billing").join(", "));
console.log("log: analytics reads", log.poll("analytics").join(", "), "(independently, same events)");
log.append("order-5");
console.log("log: billing then reads", log.poll("billing").join(", "));
log.seek("analytics", 0);
console.log("log: analytics replays", log.poll("analytics").length, "events from offset 0");

import java.util.*;

public class Main {
    static class Queue {
        // A work queue: each message goes to one worker; unacked messages come back; repeated failures go to a DLQ.
        final Deque<String> ready = new ArrayDeque<>();
        final List<String> dlq = new ArrayList<>();
        final Map<String, Integer> tries = new HashMap<>();
        final int maxTries;
        Queue(int maxTries) { this.maxTries = maxTries; }
        void send(String msg) { ready.addLast(msg); }
        String receive() { return ready.pollFirst(); }
        void nack(String msg) {                // crash or error: the visibility timeout expires and it comes back
            int n = tries.merge(msg, 1, Integer::sum);
            if (n >= maxTries) dlq.add(msg); else ready.addLast(msg);
        }
    }

    static class Log {
        // An event log: messages stay; each consumer group keeps its own offset and can rewind.
        final List<String> entries = new ArrayList<>();
        final Map<String, Integer> offsets = new HashMap<>();
        void append(String msg) { entries.add(msg); }
        List<String> poll(String group) {
            int start = offsets.getOrDefault(group, 0);
            offsets.put(group, entries.size());
            return new ArrayList<>(entries.subList(start, entries.size()));
        }
        void seek(String group, int offset) { offsets.put(group, offset); }
    }

    public static void main(String[] args) {
        List<String> orders = List.of("order-1", "order-2", "poison", "order-4");
        Queue q = new Queue(3);
        for (String o : orders) q.send(o);
        List<String> done = new ArrayList<>();
        boolean crashedOnce = false;
        String msg;
        while ((msg = q.receive()) != null) {
            if (msg.equals("poison") || (msg.equals("order-2") && !crashedOnce)) {
                crashedOnce = crashedOnce || msg.equals("order-2");
                q.nack(msg);                       // order-2 fails once (a worker crash); poison fails every time
            } else done.add(msg);
        }
        System.out.println("queue: processed " + String.join(", ", done) + " | dead-letter: " + String.join(", ", q.dlq) + " | poison tried " + q.tries.get("poison") + " times");

        Log log = new Log();
        for (String o : orders) log.append(o);
        System.out.println("log: billing reads " + String.join(", ", log.poll("billing")));
        System.out.println("log: analytics reads " + String.join(", ", log.poll("analytics")) + " (independently, same events)");
        log.append("order-5");
        System.out.println("log: billing then reads " + String.join(", ", log.poll("billing")));
        log.seek("analytics", 0);
        System.out.println("log: analytics replays " + log.poll("analytics").size() + " events from offset 0");
    }
}