10. Pub-Sub / Message Queue Advanced + Concurrency

interface Subscriber {
  void onMessage(Message message);
}

interface OverflowPolicy {
  void onOverflow(Subscription subscription, Message incoming);
}

class Broker {
  Topic createTopic(String name);
  void publish(String topicName, Message message);
  Subscription subscribe(String topicName, Subscriber subscriber, OverflowPolicy policy);
  void unsubscribe(Subscription subscription);
}

class Subscription {
  void ack(String messageId);
  void nack(String messageId);   // triggers redelivery, up to a retry limit, then dead-letter
}

interface Subscriber {
  void onMessage(Message message);
}

interface OverflowPolicy {
  void onOverflow(Subscription subscription, Message incoming);
}

class Broker {
  Topic createTopic(String name);
  void publish(String topicName, Message message);
  Subscription subscribe(String topicName, Subscriber subscriber, OverflowPolicy policy);
  void unsubscribe(Subscription subscription);
}

class Subscription {
  void ack(String messageId);
  void nack(String messageId);   // triggers redelivery, up to a retry limit, then dead-letter
}

class Subscriber {
public:
  virtual void onMessage(const Message& message) = 0;
};

class OverflowPolicy {
public:
  virtual void onOverflow(std::shared_ptr<Subscription> subscription, const Message& incoming) = 0;
};

class Broker {
public:
  std::shared_ptr<Topic> createTopic(const std::string& name);
  void publish(const std::string& topicName, const Message& message);
  std::shared_ptr<Subscription> subscribe(const std::string& topicName, std::shared_ptr<Subscriber> subscriber,
                                           std::shared_ptr<OverflowPolicy> policy);
  void unsubscribe(std::shared_ptr<Subscription> subscription);
};

class Subscription {
public:
  void ack(const std::string& messageId);
  void nack(const std::string& messageId);   // triggers redelivery, up to a retry limit, then dead-letter
};

class Subscriber {
  onMessage(message) { throw new Error('not implemented'); }
}

class OverflowPolicy {
  onOverflow(subscription, incoming) { throw new Error('not implemented'); }
}

class Broker {
  createTopic(name) {}
  publish(topicName, message) {}
  subscribe(topicName, subscriber, policy) {}
  unsubscribe(subscription) {}
}

class Subscription {
  ack(messageId) {}
  nack(messageId) {}   // triggers redelivery, up to a retry limit, then dead-letter
}

interface Subscriber {
  onMessage(message: Message): void;
}

interface OverflowPolicy {
  onOverflow(subscription: Subscription, incoming: Message): void;
}

class Broker {
  createTopic(name: string): Topic { return null as unknown as Topic; }
  publish(topicName: string, message: Message): void {}
  subscribe(topicName: string, subscriber: Subscriber, policy: OverflowPolicy): Subscription {
    return null as unknown as Subscription;
  }
  unsubscribe(subscription: Subscription): void {}
}

class Subscription {
  ack(messageId: string): void {}
  nack(messageId: string): void {}   // triggers redelivery, up to a retry limit, then dead-letter
}

Topic.publish(message):
  for subscription in this.subscriptions:
    if not subscription.queue.tryEnqueue(message):     // bounded queue full
      subscription.overflowPolicy.onOverflow(subscription, message)

Subscription.consumeLoop():                             // one per subscriber, background thread
  while running:
    message = this.queue.take()
    try:
      this.subscriber.onMessage(message)                // must be idempotent — may be a redelivery
      this.ack(message.id)
    catch:
      this.nack(message.id)

Subscription.nack(messageId):
  attempts = this.retryCount.increment(messageId)
  if attempts > MAX_RETRIES:
    this.deadLetterQueue.add(message, reason="retry exhausted")
  else:
    this.queue.tryEnqueue(message)   // redeliver — subscriber will see this message again

void publish(Message message) {
  for (Subscription subscription : this.subscriptions) {
    if (!subscription.getQueue().tryEnqueue(message)) {      // bounded queue full
      subscription.getOverflowPolicy().onOverflow(subscription, message);
    }
  }
}

void consumeLoop() {                                          // one per subscriber, background thread
  while (running) {
    Message message = this.queue.take();
    try {
      this.subscriber.onMessage(message);                     // must be idempotent — may be a redelivery
      this.ack(message.getId());
    } catch (Exception e) {
      this.nack(message.getId());
    }
  }
}

void nack(String messageId) {
  int attempts = this.retryCount.increment(messageId);
  if (attempts > MAX_RETRIES) {
    this.deadLetterQueue.add(message, "retry exhausted");
  } else {
    this.queue.tryEnqueue(message);   // redeliver — subscriber will see this message again
  }
}

void Topic::publish(const Message& message) {
  for (auto& subscription : subscriptions_) {
    if (!subscription->queue().tryEnqueue(message)) {          // bounded queue full
      subscription->overflowPolicy()->onOverflow(subscription, message);
    }
  }
}

void Subscription::consumeLoop() {                              // one per subscriber, background thread
  while (running_) {
    auto message = queue_.take();
    try {
      subscriber_->onMessage(message);                          // must be idempotent — may be a redelivery
      ack(message.id);
    } catch (...) {
      nack(message.id);
    }
  }
}

void Subscription::nack(const std::string& messageId) {
  int attempts = retryCount_.increment(messageId);
  if (attempts > MAX_RETRIES) {
    deadLetterQueue_.add(message, "retry exhausted");
  } else {
    queue_.tryEnqueue(message);   // redeliver — subscriber will see this message again
  }
}

publish(message) {
  for (const subscription of this.subscriptions) {
    if (!subscription.queue.tryEnqueue(message)) {              // bounded queue full
      subscription.overflowPolicy.onOverflow(subscription, message);
    }
  }
}

async consumeLoop() {                                           // one per subscriber, background thread
  while (this.running) {
    const message = await this.queue.take();
    try {
      this.subscriber.onMessage(message);                       // must be idempotent — may be a redelivery
      this.ack(message.id);
    } catch (e) {
      this.nack(message.id);
    }
  }
}

nack(messageId) {
  const attempts = this.retryCount.increment(messageId);
  if (attempts > MAX_RETRIES) {
    this.deadLetterQueue.add(message, "retry exhausted");
  } else {
    this.queue.tryEnqueue(message);   // redeliver — subscriber will see this message again
  }
}

publish(message: Message): void {
  for (const subscription of this.subscriptions) {
    if (!subscription.queue.tryEnqueue(message)) {               // bounded queue full
      subscription.overflowPolicy.onOverflow(subscription, message);
    }
  }
}

async consumeLoop(): Promise<void> {                             // one per subscriber, background thread
  while (this.running) {
    const message: Message = await this.queue.take();
    try {
      this.subscriber.onMessage(message);                        // must be idempotent — may be a redelivery
      this.ack(message.id);
    } catch (e) {
      this.nack(message.id);
    }
  }
}

nack(messageId: string): void {
  const attempts = this.retryCount.increment(messageId);
  if (attempts > MAX_RETRIES) {
    this.deadLetterQueue.add(message, "retry exhausted");
  } else {
    this.queue.tryEnqueue(message);    // redeliver — subscriber will see this message again
  }
}

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Pub-Sub / Message Queue — full class-diagram implementation.
 *
 * Broker owns Topics; a Topic owns Subscriptions (one per Subscriber); each Subscription
 * has its own bounded queue + background consumer thread so a slow subscriber can never
 * block the publisher or other subscribers (decoupling publish() from consumption).
 */
public class p10 {

    // ---------- Message ----------
    /** Immutable unit of data plus metadata. */
    static final class Message {
        final String id;
        final String payload;
        final String partitionKey;
        final long timestamp;

        Message(String id, String payload, String partitionKey) {
            this.id = id;
            this.payload = payload;
            this.partitionKey = partitionKey;
            this.timestamp = System.currentTimeMillis();
        }

        @Override
        public String toString() {
            return "Message{id=" + id + ", payload=" + payload + "}";
        }
    }

    // ---------- Subscriber (interface) ----------
    /** Business logic for handling one message. Must be idempotent (at-least-once delivery). */
    interface Subscriber {
        void onMessage(Message message) throws Exception;
    }

    // ---------- OverflowPolicy (Strategy interface) ----------
    /** What happens when a subscriber's queue is full. */
    interface OverflowPolicy {
        void onOverflow(Subscription subscription, Message incoming);
    }

    /** Drop the oldest queued message to make room for the incoming one. */
    static final class DropOldest implements OverflowPolicy {
        @Override
        public void onOverflow(Subscription subscription, Message incoming) {
            subscription.queue.poll();               // discard oldest
            subscription.queue.offer(incoming);       // make room for the new message
        }
    }

    /** Block the publishing thread until space is available (backpressure). */
    static final class BlockPublisher implements OverflowPolicy {
        @Override
        public void onOverflow(Subscription subscription, Message incoming) {
            try {
                subscription.queue.put(incoming);      // blocks caller thread until room frees up
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /** Simply drop the new incoming message, keeping what's already queued. */
    static final class DropNewest implements OverflowPolicy {
        @Override
        public void onOverflow(Subscription subscription, Message incoming) {
            // intentionally no-op: newest message is dropped
        }
    }

    // ---------- DeadLetterQueue ----------
    /** Holds messages that exhausted their retry budget. Reprocessing is manual/operator-triggered. */
    static final class DeadLetterQueue {
        private final List<String> entries = new CopyOnWriteArrayList<>();

        void add(Message message, String reason) {
            entries.add(message.id + " :: " + reason);
        }

        List<String> entries() {
            return entries;
        }
    }

    // ---------- Subscription ----------
    /** Per-subscriber delivery state on a topic: bounded queue, ack/retry tracking, background consumer. */
    static final class Subscription {
        private static final int MAX_RETRIES = 3;
        private static final int CAPACITY = 16;

        final String id = UUID.randomUUID().toString();
        final BlockingQueue<Message> queue = new ArrayBlockingQueue<>(CAPACITY);
        private final Subscriber subscriber;
        private final OverflowPolicy overflowPolicy;
        private final DeadLetterQueue deadLetterQueue;
        private final Map<String, Message> inFlight = new ConcurrentHashMap<>();
        private final Map<String, AtomicInteger> retryCount = new ConcurrentHashMap<>();
        private volatile boolean running = true;
        private Thread consumerThread;

        Subscription(Subscriber subscriber, OverflowPolicy overflowPolicy, DeadLetterQueue dlq) {
            this.subscriber = subscriber;
            this.overflowPolicy = overflowPolicy;
            this.deadLetterQueue = dlq;
        }

        /** Try to enqueue; on a full queue delegate to the overflow policy. */
        void enqueue(Message message) {
            if (!queue.offer(message)) {
                overflowPolicy.onOverflow(this, message);
            }
        }

        void start() {
            consumerThread = new Thread(this::consumeLoop, "sub-" + id.substring(0, 8));
            consumerThread.setDaemon(true);
            consumerThread.start();
        }

        void stop() {
            running = false;
            if (consumerThread != null) consumerThread.interrupt();
        }

        /** One consumer loop per subscriber, running on its own background thread. */
        private void consumeLoop() {
            while (running) {
                try {
                    Message message = queue.poll(200, TimeUnit.MILLISECONDS);
                    if (message == null) continue;
                    inFlight.put(message.id, message);
                    try {
                        subscriber.onMessage(message);   // must be idempotent -- may be a redelivery
                        ack(message.id);
                    } catch (Exception e) {
                        nack(message.id);
                    }
                } catch (InterruptedException ie) {
                    if (!running) break;
                }
            }
        }

        void ack(String messageId) {
            inFlight.remove(messageId);
            retryCount.remove(messageId);
        }

        void nack(String messageId) {
            Message message = inFlight.remove(messageId);
            if (message == null) return;
            int attempts = retryCount.computeIfAbsent(messageId, k -> new AtomicInteger(0))
                    .incrementAndGet();
            if (attempts > MAX_RETRIES) {
                deadLetterQueue.add(message, "retry exhausted");
                retryCount.remove(messageId);
            } else {
                enqueue(message);   // redeliver -- subscriber will see this message again
            }
        }
    }

    // ---------- Topic ----------
    /** A named channel; owns its subscriber list and fans out on publish. */
    static final class Topic {
        final String name;
        private final List<Subscription> subscriptions = new CopyOnWriteArrayList<>();

        Topic(String name) {
            this.name = name;
        }

        Subscription subscribe(Subscriber subscriber, OverflowPolicy policy, DeadLetterQueue dlq) {
            Subscription subscription = new Subscription(subscriber, policy, dlq);
            subscriptions.add(subscription);
            subscription.start();
            return subscription;
        }

        void unsubscribe(Subscription subscription) {
            subscriptions.remove(subscription);
            subscription.stop();
        }

        /** Fan out to every subscription's own bounded queue -- never calls the subscriber inline. */
        void publish(Message message) {
            for (Subscription subscription : subscriptions) {
                subscription.enqueue(message);
            }
        }
    }

    // ---------- Broker (facade) ----------
    /** Owns all topics; top-level publish/subscribe entry point. */
    static final class Broker {
        private final Map<String, Topic> topics = new ConcurrentHashMap<>();
        private final DeadLetterQueue deadLetterQueue = new DeadLetterQueue();

        Topic createTopic(String name) {
            return topics.computeIfAbsent(name, Topic::new);
        }

        void publish(String topicName, Message message) {
            Topic topic = topics.get(topicName);
            if (topic == null) throw new IllegalArgumentException("No such topic: " + topicName);
            topic.publish(message);
        }

        Subscription subscribe(String topicName, Subscriber subscriber, OverflowPolicy policy) {
            Topic topic = topics.get(topicName);
            if (topic == null) throw new IllegalArgumentException("No such topic: " + topicName);
            return topic.subscribe(subscriber, policy, deadLetterQueue);
        }

        void unsubscribe(String topicName, Subscription subscription) {
            Topic topic = topics.get(topicName);
            if (topic != null) topic.unsubscribe(subscription);
        }

        DeadLetterQueue deadLetterQueue() {
            return deadLetterQueue;
        }
    }

    // ---------- Demo ----------
    public static void main(String[] args) throws InterruptedException {
        Broker broker = new Broker();
        broker.createTopic("orders");

        List<String> fastReceived = new CopyOnWriteArrayList<>();
        Subscriber fastSubscriber = msg -> fastReceived.add(msg.payload);

        // Always fails for one specific message id to deterministically exercise
        // the retry -> dead-letter path while other messages succeed normally.
        Subscriber flakySubscriber = msg -> {
            if (msg.id.equals("m2")) {
                throw new RuntimeException("permanent failure for m2");
            }
        };

        broker.subscribe("orders", fastSubscriber, new DropOldest());
        broker.subscribe("orders", flakySubscriber, new DropOldest());

        for (int i = 0; i < 5; i++) {
            broker.publish("orders", new Message("m" + i, "order-" + i, "p0"));
        }

        Thread.sleep(1000);

        System.out.println("Fast subscriber received: " + fastReceived);
        System.out.println("Dead letters: " + broker.deadLetterQueue().entries());
    }
}

// Pub-Sub / Message Queue -- full class-diagram implementation.
//
// Broker owns Topics; a Topic owns Subscriptions (one per Subscriber); each Subscription
// has its own bounded queue + background consumer thread so a slow subscriber can never
// block the publisher or other subscribers.

#include <iostream>
#include <string>
#include <vector>
#include <map>
#include <unordered_map>
#include <deque>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <chrono>
#include <functional>
#include <sstream>
#include <algorithm>
#include <stdexcept>

// ---------- Message ----------
struct Message {
    std::string id;
    std::string payload;
    std::string partitionKey;
    long long timestamp;

    Message(std::string id_, std::string payload_, std::string partitionKey_)
        : id(std::move(id_)), payload(std::move(payload_)), partitionKey(std::move(partitionKey_)) {
        timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
    }
};

// ---------- Subscriber (interface) ----------
// Business logic for handling one message. Must be idempotent (at-least-once delivery).
struct Subscriber {
    virtual ~Subscriber() = default;
    virtual void onMessage(const Message& message) = 0; // throws on failure
};

class Subscription; // fwd decl

// ---------- OverflowPolicy (Strategy interface) ----------
struct OverflowPolicy {
    virtual ~OverflowPolicy() = default;
    virtual void onOverflow(Subscription& subscription, const Message& incoming) = 0;
};

// ---------- DeadLetterQueue ----------
class DeadLetterQueue {
    std::mutex mtx;
    std::vector<std::string> entries_;
public:
    void add(const Message& message, const std::string& reason) {
        std::lock_guard<std::mutex> lock(mtx);
        entries_.push_back(message.id + " :: " + reason);
    }
    std::vector<std::string> entries() {
        std::lock_guard<std::mutex> lock(mtx);
        return entries_;
    }
};

// ---------- Subscription ----------
// Per-subscriber delivery state on a topic: bounded queue, ack/retry tracking, background consumer.
class Subscription {
public:
    static const size_t CAPACITY = 16;
    static const int MAX_RETRIES = 3;

    std::string id;
    std::deque<Message> queue;
    std::mutex queueMutex;
    std::condition_variable queueCv;

    Subscription(std::shared_ptr<Subscriber> subscriber,
                 std::shared_ptr<OverflowPolicy> overflowPolicy,
                 std::shared_ptr<DeadLetterQueue> dlq,
                 std::string subId)
        : id(std::move(subId)), subscriber_(std::move(subscriber)),
          overflowPolicy_(std::move(overflowPolicy)), deadLetterQueue_(std::move(dlq)) {}

    ~Subscription() { stop(); }

    // Try to enqueue; on a full queue delegate to the overflow policy.
    void enqueue(const Message& message) {
        std::unique_lock<std::mutex> lock(queueMutex);
        if (queue.size() < CAPACITY) {
            queue.push_back(message);
            lock.unlock();
            queueCv.notify_one();
        } else {
            lock.unlock();
            overflowPolicy_->onOverflow(*this, message);   // policy may itself re-lock and enqueue
        }
    }

    // Raw push used only by overflow policies that already decided to bypass capacity checks.
    void forcePush(const Message& message) {
        std::unique_lock<std::mutex> lock(queueMutex);
        queue.push_back(message);
        lock.unlock();
        queueCv.notify_one();
    }

    void popOldest() {
        std::lock_guard<std::mutex> lock(queueMutex);
        if (!queue.empty()) queue.pop_front();
    }

    void start() {
        running_ = true;
        thread_ = std::thread(&Subscription::consumeLoop, this);
    }

    void stop() {
        if (!running_) return;
        running_ = false;
        queueCv.notify_all();
        if (thread_.joinable()) thread_.join();
    }

    void ack(const std::string& messageId) {
        std::lock_guard<std::mutex> lock(stateMutex_);
        inFlight_.erase(messageId);
        retryCount_.erase(messageId);
    }

    void nack(const std::string& messageId) {
        std::lock_guard<std::mutex> lock(stateMutex_);
        auto it = inFlight_.find(messageId);
        if (it == inFlight_.end()) return;
        Message message = it->second;
        inFlight_.erase(it);
        int attempts = ++retryCount_[messageId];
        if (attempts > MAX_RETRIES) {
            deadLetterQueue_->add(message, "retry exhausted");
            retryCount_.erase(messageId);
        } else {
            enqueue(message);   // redeliver -- subscriber will see this message again
        }
    }

private:
    std::shared_ptr<Subscriber> subscriber_;
    std::shared_ptr<OverflowPolicy> overflowPolicy_;
    std::shared_ptr<DeadLetterQueue> deadLetterQueue_;
    std::atomic<bool> running_{false};
    std::thread thread_;

    std::mutex stateMutex_;
    std::unordered_map<std::string, Message> inFlight_;
    std::unordered_map<std::string, int> retryCount_;

    // One consumer loop per subscriber, running on its own background thread.
    void consumeLoop() {
        while (running_) {
            Message message("", "", "");
            bool got = false;
            {
                std::unique_lock<std::mutex> lock(queueMutex);
                queueCv.wait_for(lock, std::chrono::milliseconds(200),
                                  [this] { return !queue.empty() || !running_; });
                if (!queue.empty()) {
                    message = queue.front();
                    queue.pop_front();
                    got = true;
                }
            }
            if (!got) continue;
            {
                std::lock_guard<std::mutex> lock(stateMutex_);
                inFlight_.emplace(message.id, message);
            }
            try {
                subscriber_->onMessage(message);   // must be idempotent -- may be a redelivery
                ack(message.id);
            } catch (...) {
                nack(message.id);
            }
        }
    }
};

// Drop the oldest queued message to make room for the incoming one.
struct DropOldest : OverflowPolicy {
    void onOverflow(Subscription& subscription, const Message& incoming) override {
        subscription.popOldest();
        subscription.forcePush(incoming);
    }
};

// Block the publishing thread until space is available (backpressure).
struct BlockPublisher : OverflowPolicy {
    void onOverflow(Subscription& subscription, const Message& incoming) override {
        std::unique_lock<std::mutex> lock(subscription.queueMutex);
        subscription.queueCv.wait(lock, [&] { return subscription.queue.size() < Subscription::CAPACITY; });
        subscription.queue.push_back(incoming);
        lock.unlock();
        subscription.queueCv.notify_one();
    }
};

// Simply drop the new incoming message, keeping what's already queued.
struct DropNewest : OverflowPolicy {
    void onOverflow(Subscription& subscription, const Message& incoming) override {
        (void)subscription; (void)incoming; // intentionally no-op
    }
};

// ---------- Topic ----------
// A named channel; owns its subscriber list and fans out on publish.
class Topic {
public:
    std::string name;
    explicit Topic(std::string name_) : name(std::move(name_)) {}

    std::shared_ptr<Subscription> subscribe(std::shared_ptr<Subscriber> subscriber,
                                             std::shared_ptr<OverflowPolicy> policy,
                                             std::shared_ptr<DeadLetterQueue> dlq) {
        std::lock_guard<std::mutex> lock(mtx_);
        auto sub = std::make_shared<Subscription>(subscriber, policy, dlq,
                                                    "sub" + std::to_string(nextId_++));
        subscriptions_.push_back(sub);
        sub->start();
        return sub;
    }

    void unsubscribe(const std::shared_ptr<Subscription>& subscription) {
        std::lock_guard<std::mutex> lock(mtx_);
        subscription->stop();
        subscriptions_.erase(std::remove(subscriptions_.begin(), subscriptions_.end(), subscription),
                              subscriptions_.end());
    }

    // Fan out to every subscription's own bounded queue -- never calls the subscriber inline.
    void publish(const Message& message) {
        std::vector<std::shared_ptr<Subscription>> snapshot;
        {
            std::lock_guard<std::mutex> lock(mtx_);
            snapshot = subscriptions_;
        }
        for (auto& sub : snapshot) sub->enqueue(message);
    }

private:
    std::mutex mtx_;
    std::vector<std::shared_ptr<Subscription>> subscriptions_;
    int nextId_ = 0;
};

// ---------- Broker (facade) ----------
// Owns all topics; top-level publish/subscribe entry point.
class Broker {
public:
    Broker() : deadLetterQueue_(std::make_shared<DeadLetterQueue>()) {}

    std::shared_ptr<Topic> createTopic(const std::string& name) {
        std::lock_guard<std::mutex> lock(mtx_);
        auto it = topics_.find(name);
        if (it != topics_.end()) return it->second;
        auto topic = std::make_shared<Topic>(name);
        topics_[name] = topic;
        return topic;
    }

    void publish(const std::string& topicName, const Message& message) {
        auto topic = getTopic(topicName);
        topic->publish(message);
    }

    std::shared_ptr<Subscription> subscribe(const std::string& topicName,
                                             std::shared_ptr<Subscriber> subscriber,
                                             std::shared_ptr<OverflowPolicy> policy) {
        auto topic = getTopic(topicName);
        return topic->subscribe(subscriber, policy, deadLetterQueue_);
    }

    void unsubscribe(const std::string& topicName, const std::shared_ptr<Subscription>& subscription) {
        auto topic = getTopic(topicName);
        topic->unsubscribe(subscription);
    }

    std::shared_ptr<DeadLetterQueue> deadLetterQueue() { return deadLetterQueue_; }

private:
    std::mutex mtx_;
    std::map<std::string, std::shared_ptr<Topic>> topics_;
    std::shared_ptr<DeadLetterQueue> deadLetterQueue_;

    std::shared_ptr<Topic> getTopic(const std::string& name) {
        std::lock_guard<std::mutex> lock(mtx_);
        auto it = topics_.find(name);
        if (it == topics_.end()) throw std::invalid_argument("No such topic: " + name);
        return it->second;
    }
};

// ---------- Demo subscribers ----------
struct FastSubscriber : Subscriber {
    std::mutex mtx;
    std::vector<std::string> received;
    void onMessage(const Message& message) override {
        std::lock_guard<std::mutex> lock(mtx);
        received.push_back(message.payload);
    }
};

// Always fails for one specific message id to deterministically exercise
// the retry -> dead-letter path while other messages succeed normally.
struct FlakySubscriber : Subscriber {
    void onMessage(const Message& message) override {
        if (message.id == "m2") throw std::runtime_error("permanent failure for m2");
    }
};

int main() {
    Broker broker;
    broker.createTopic("orders");

    auto fast = std::make_shared<FastSubscriber>();
    auto flaky = std::make_shared<FlakySubscriber>();

    broker.subscribe("orders", fast, std::make_shared<DropOldest>());
    broker.subscribe("orders", flaky, std::make_shared<DropOldest>());

    for (int i = 0; i < 5; i++) {
        broker.publish("orders", Message("m" + std::to_string(i), "order-" + std::to_string(i), "p0"));
    }

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::cout << "Fast subscriber received: ";
    for (auto& p : fast->received) std::cout << p << " ";
    std::cout << "\n";

    std::cout << "Dead letters: ";
    for (auto& e : broker.deadLetterQueue()->entries()) std::cout << e << " | ";
    std::cout << "\n";

    return 0;
}

'use strict';
/**
 * Pub-Sub / Message Queue -- full class-diagram implementation.
 *
 * Broker owns Topics; a Topic owns Subscriptions (one per Subscriber); each Subscription
 * has its own bounded queue + async consumption loop so a slow subscriber can never
 * block the publisher or other subscribers (decoupling publish() from consumption).
 *
 * Node is single-threaded, so "concurrency" here means interleaved async consumer loops
 * (via setImmediate) rather than OS threads -- the same bounded-queue-per-subscriber
 * design guards against one subscriber starving another, and enqueue/ack/nack are each
 * atomic within a single synchronous turn of the event loop.
 */

const { randomUUID } = require('crypto');

// ---------- Message ----------
class Message {
  constructor(id, payload, partitionKey) {
    this.id = id;
    this.payload = payload;
    this.partitionKey = partitionKey;
    this.timestamp = Date.now();
  }
}

// ---------- OverflowPolicy (Strategy interface) ----------
class OverflowPolicy {
  onOverflow(_subscription, _incoming) {
    throw new Error('not implemented');
  }
}

class DropOldest extends OverflowPolicy {
  onOverflow(subscription, incoming) {
    subscription.queue.shift();     // discard oldest
    subscription.queue.push(incoming);
  }
}

class DropNewest extends OverflowPolicy {
  onOverflow(_subscription, _incoming) {
    // intentionally no-op: newest message is dropped
  }
}

class BlockPublisher extends OverflowPolicy {
  onOverflow(subscription, incoming) {
    // No real OS blocking available in single-threaded JS; approximate
    // backpressure by force-pushing past capacity (publisher "waits its turn"
    // conceptually -- the queue will drain before growing further).
    subscription.queue.push(incoming);
  }
}

// ---------- DeadLetterQueue ----------
class DeadLetterQueue {
  constructor() {
    this._entries = [];
  }
  add(message, reason) {
    this._entries.push(`${message.id} :: ${reason}`);
  }
  entries() {
    return this._entries;
  }
}

// ---------- Subscription ----------
const CAPACITY = 16;
const MAX_RETRIES = 3;

class Subscription {
  constructor(subscriber, overflowPolicy, deadLetterQueue) {
    this.id = randomUUID();
    this.subscriber = subscriber;       // Subscriber: { onMessage(message) }
    this.overflowPolicy = overflowPolicy;
    this.deadLetterQueue = deadLetterQueue;
    this.queue = [];
    this.inFlight = new Map();
    this.retryCount = new Map();
    this.running = false;
    this._timer = null;
  }

  // Try to enqueue; on a full queue delegate to the overflow policy.
  enqueue(message) {
    if (this.queue.length < CAPACITY) {
      this.queue.push(message);
    } else {
      this.overflowPolicy.onOverflow(this, message);
    }
  }

  start() {
    this.running = true;
    this._scheduleNext();
  }

  stop() {
    this.running = false;
    if (this._timer) clearImmediate(this._timer);
  }

  _scheduleNext() {
    if (!this.running) return;
    this._timer = setImmediate(() => this._consumeOne());
  }

  // One "tick" of the consumer loop -- pulls one message and processes it.
  async _consumeOne() {
    if (!this.running) return;
    const message = this.queue.shift();
    if (!message) {
      this._scheduleNext();
      return;
    }
    this.inFlight.set(message.id, message);
    try {
      await this.subscriber.onMessage(message); // must be idempotent -- may be a redelivery
      this.ack(message.id);
    } catch (_err) {
      this.nack(message.id);
    }
    this._scheduleNext();
  }

  ack(messageId) {
    this.inFlight.delete(messageId);
    this.retryCount.delete(messageId);
  }

  nack(messageId) {
    const message = this.inFlight.get(messageId);
    this.inFlight.delete(messageId);
    if (!message) return;
    const attempts = (this.retryCount.get(messageId) || 0) + 1;
    this.retryCount.set(messageId, attempts);
    if (attempts > MAX_RETRIES) {
      this.deadLetterQueue.add(message, 'retry exhausted');
      this.retryCount.delete(messageId);
    } else {
      this.enqueue(message); // redeliver -- subscriber will see this message again
    }
  }
}

// ---------- Topic ----------
class Topic {
  constructor(name) {
    this.name = name;
    this.subscriptions = [];
  }

  subscribe(subscriber, policy, dlq) {
    const subscription = new Subscription(subscriber, policy, dlq);
    this.subscriptions.push(subscription);
    subscription.start();
    return subscription;
  }

  unsubscribe(subscription) {
    subscription.stop();
    this.subscriptions = this.subscriptions.filter((s) => s !== subscription);
  }

  // Fan out to every subscription's own bounded queue -- never calls the subscriber inline.
  publish(message) {
    for (const subscription of this.subscriptions) {
      subscription.enqueue(message);
    }
  }
}

// ---------- Broker (facade) ----------
class Broker {
  constructor() {
    this.topics = new Map();
    this.deadLetterQueue = new DeadLetterQueue();
  }

  createTopic(name) {
    if (!this.topics.has(name)) this.topics.set(name, new Topic(name));
    return this.topics.get(name);
  }

  _getTopic(name) {
    const topic = this.topics.get(name);
    if (!topic) throw new Error(`No such topic: ${name}`);
    return topic;
  }

  publish(topicName, message) {
    this._getTopic(topicName).publish(message);
  }

  subscribe(topicName, subscriber, policy) {
    return this._getTopic(topicName).subscribe(subscriber, policy, this.deadLetterQueue);
  }

  unsubscribe(topicName, subscription) {
    this._getTopic(topicName).unsubscribe(subscription);
  }
}

// ---------- Demo ----------
function main() {
  const broker = new Broker();
  broker.createTopic('orders');

  const fastReceived = [];
  const fastSubscriber = { onMessage: (msg) => { fastReceived.push(msg.payload); } };

  // Always fails for one specific message id to deterministically exercise
  // the retry -> dead-letter path while other messages succeed normally.
  const flakySubscriber = {
    onMessage: (msg) => {
      if (msg.id === 'm2') throw new Error('permanent failure for m2');
    },
  };

  broker.subscribe('orders', fastSubscriber, new DropOldest());
  broker.subscribe('orders', flakySubscriber, new DropOldest());

  for (let i = 0; i < 5; i++) {
    broker.publish('orders', new Message(`m${i}`, `order-${i}`, 'p0'));
  }

  setTimeout(() => {
    console.log('Fast subscriber received:', fastReceived);
    console.log('Dead letters:', broker.deadLetterQueue.entries());
  }, 500);
}

if (require.main === module) {
  main();
}

module.exports = {
  Message, Topic, Broker, Subscription,
  DropOldest, DropNewest, BlockPublisher, DeadLetterQueue,
};

/**
 * Pub-Sub / Message Queue -- full class-diagram implementation.
 *
 * Broker owns Topics; a Topic owns Subscriptions (one per Subscriber); each Subscription
 * has its own bounded queue + async consumption loop so a slow subscriber can never
 * block the publisher or other subscribers (decoupling publish() from consumption).
 */

import { randomUUID } from 'crypto';

// ---------- Message ----------
class Message {
  readonly timestamp: number;
  constructor(
    readonly id: string,
    readonly payload: string,
    readonly partitionKey: string,
  ) {
    this.timestamp = Date.now();
  }
}

// ---------- Subscriber (interface) ----------
interface Subscriber {
  onMessage(message: Message): void | Promise<void>;
}

// ---------- OverflowPolicy (Strategy interface) ----------
interface OverflowPolicy {
  onOverflow(subscription: Subscription, incoming: Message): void;
}

class DropOldest implements OverflowPolicy {
  onOverflow(subscription: Subscription, incoming: Message): void {
    subscription.queue.shift();
    subscription.queue.push(incoming);
  }
}

class DropNewest implements OverflowPolicy {
  onOverflow(_subscription: Subscription, _incoming: Message): void {
    // intentionally no-op: newest message is dropped
  }
}

class BlockPublisher implements OverflowPolicy {
  onOverflow(subscription: Subscription, incoming: Message): void {
    // Single-threaded JS has no real blocking primitive; approximate
    // backpressure by force-pushing -- the queue drains before growing further.
    subscription.queue.push(incoming);
  }
}

// ---------- DeadLetterQueue ----------
class DeadLetterQueue {
  private readonly _entries: string[] = [];

  add(message: Message, reason: string): void {
    this._entries.push(`${message.id} :: ${reason}`);
  }

  entries(): string[] {
    return this._entries;
  }
}

// ---------- Subscription ----------
const CAPACITY = 16;
const MAX_RETRIES = 3;

class Subscription {
  readonly id: string = randomUUID();
  readonly queue: Message[] = [];
  private readonly inFlight = new Map<string, Message>();
  private readonly retryCount = new Map<string, number>();
  private running = false;

  constructor(
    private readonly subscriber: Subscriber,
    private readonly overflowPolicy: OverflowPolicy,
    private readonly deadLetterQueue: DeadLetterQueue,
  ) {}

  /** Try to enqueue; on a full queue delegate to the overflow policy. */
  enqueue(message: Message): void {
    if (this.queue.length < CAPACITY) {
      this.queue.push(message);
    } else {
      this.overflowPolicy.onOverflow(this, message);
    }
  }

  start(): void {
    this.running = true;
    this.scheduleNext();
  }

  stop(): void {
    this.running = false;
  }

  private scheduleNext(): void {
    if (!this.running) return;
    setImmediate(() => this.consumeOne());
  }

  /** One "tick" of the consumer loop -- pulls one message and processes it. */
  private async consumeOne(): Promise<void> {
    if (!this.running) return;
    const message = this.queue.shift();
    if (!message) {
      this.scheduleNext();
      return;
    }
    this.inFlight.set(message.id, message);
    try {
      await this.subscriber.onMessage(message); // must be idempotent -- may be a redelivery
      this.ack(message.id);
    } catch {
      this.nack(message.id);
    }
    this.scheduleNext();
  }

  ack(messageId: string): void {
    this.inFlight.delete(messageId);
    this.retryCount.delete(messageId);
  }

  nack(messageId: string): void {
    const message = this.inFlight.get(messageId);
    this.inFlight.delete(messageId);
    if (!message) return;
    const attempts = (this.retryCount.get(messageId) ?? 0) + 1;
    this.retryCount.set(messageId, attempts);
    if (attempts > MAX_RETRIES) {
      this.deadLetterQueue.add(message, 'retry exhausted');
      this.retryCount.delete(messageId);
    } else {
      this.enqueue(message); // redeliver -- subscriber will see this message again
    }
  }
}

// ---------- Topic ----------
class Topic {
  private subscriptions: Subscription[] = [];

  constructor(readonly name: string) {}

  subscribe(subscriber: Subscriber, policy: OverflowPolicy, dlq: DeadLetterQueue): Subscription {
    const subscription = new Subscription(subscriber, policy, dlq);
    this.subscriptions.push(subscription);
    subscription.start();
    return subscription;
  }

  unsubscribe(subscription: Subscription): void {
    subscription.stop();
    this.subscriptions = this.subscriptions.filter((s) => s !== subscription);
  }

  /** Fan out to every subscription's own bounded queue -- never calls the subscriber inline. */
  publish(message: Message): void {
    for (const subscription of this.subscriptions) {
      subscription.enqueue(message);
    }
  }
}

// ---------- Broker (facade) ----------
class Broker {
  private readonly topics = new Map<string, Topic>();
  private readonly _deadLetterQueue = new DeadLetterQueue();

  createTopic(name: string): Topic {
    if (!this.topics.has(name)) this.topics.set(name, new Topic(name));
    return this.topics.get(name)!;
  }

  private getTopic(name: string): Topic {
    const topic = this.topics.get(name);
    if (!topic) throw new Error(`No such topic: ${name}`);
    return topic;
  }

  publish(topicName: string, message: Message): void {
    this.getTopic(topicName).publish(message);
  }

  subscribe(topicName: string, subscriber: Subscriber, policy: OverflowPolicy): Subscription {
    return this.getTopic(topicName).subscribe(subscriber, policy, this._deadLetterQueue);
  }

  unsubscribe(topicName: string, subscription: Subscription): void {
    this.getTopic(topicName).unsubscribe(subscription);
  }

  deadLetterQueue(): DeadLetterQueue {
    return this._deadLetterQueue;
  }
}

// ---------- Demo ----------
function main(): void {
  const broker = new Broker();
  broker.createTopic('orders');

  const fastReceived: string[] = [];
  const fastSubscriber: Subscriber = {
    onMessage: (msg) => {
      fastReceived.push(msg.payload);
    },
  };

  // Always fails for one specific message id to deterministically exercise
  // the retry -> dead-letter path while other messages succeed normally.
  const flakySubscriber: Subscriber = {
    onMessage: (msg) => {
      if (msg.id === 'm2') throw new Error('permanent failure for m2');
    },
  };

  broker.subscribe('orders', fastSubscriber, new DropOldest());
  broker.subscribe('orders', flakySubscriber, new DropOldest());

  for (let i = 0; i < 5; i++) {
    broker.publish('orders', new Message(`m${i}`, `order-${i}`, 'p0'));
  }

  setTimeout(() => {
    console.log('Fast subscriber received:', fastReceived);
    console.log('Dead letters:', broker.deadLetterQueue().entries());
  }, 500);
}

if (require.main === module) {
  main();
}

export {
  Message, Subscriber, Topic, Broker, Subscription,
  OverflowPolicy, DropOldest, DropNewest, BlockPublisher, DeadLetterQueue,
};