MQTT QoS ও বাস্তব-জীবনের প্যাটার্ন

class MqttBroker:
    """L46-এর ব্রোকার -- এই কোড সেলে পুনরায় তৈরি করা হলো যাতে সেল স্বয়ংসম্পূর্ণ থাকে।"""

    def __init__(self):
        self.subscriptions = []
        self._next_packet_id = 0

    def subscribe(self, client_id, topic_filter):
        self.subscriptions.append((client_id, topic_filter))

    @staticmethod
    def topic_matches(topic_filter, topic):
        filter_levels = topic_filter.split("/")
        topic_levels = topic.split("/")
        i = 0
        while i < len(filter_levels):
            level = filter_levels[i]
            if level == "#":
                return True
            if i >= len(topic_levels):
                return False
            if level == "+" or level == topic_levels[i]:
                i += 1
                continue
            return False
        return i == len(topic_levels)

    def route(self, topic):
        return [cid for cid, tf in self.subscriptions if self.topic_matches(tf, topic)]

    def new_packet_id(self):
        self._next_packet_id += 1
        return self._next_packet_id


def publish_qos0(broker, topic, payload):
    log = [f"[QoS 0] ক্লায়েন্ট -> ব্রোকার: PUBLISH '{topic}' (কোনো ACK ট্র্যাকিং নেই)"]
    recipients = broker.route(topic)
    log.append(f"  ব্রোকার তাৎক্ষণিক ডেলিভার করলো -> {recipients}")
    return log


def publish_qos1(broker, topic, payload, simulate_lost_ack=False):
    pid = broker.new_packet_id()
    log = []
    deliveries = []
    attempt = 1
    acked = False
    while not acked:
        dup = attempt > 1
        log.append(f"[QoS 1] ক্লায়েন্ট -> ব্রোকার: PUBLISH(id={pid}, DUP={dup}) চেষ্টা #{attempt}")
        recipients = broker.route(topic)
        deliveries.append(recipients)
        log.append(f"  ব্রোকার সাবস্ক্রাইবারদের ডেলিভার করলো -> {recipients}")
        if simulate_lost_ack and attempt == 1:
            log.append(f"  ব্রোকার -> ক্লায়েন্ট: PUBACK(id={pid}) পাঠানো হলো, কিন্তু নেটওয়ার্কে হারিয়ে গেলো (সিমুলেটেড)")
            log.append("  ক্লায়েন্ট টাইম-আউটের পর ACK না পেয়ে পুনরায় পাঠাবে")
            attempt += 1
            continue
        log.append(f"  ব্রোকার -> ক্লায়েন্ট: PUBACK(id={pid}) প্রাপ্ত -> সফল")
        acked = True
    log.append(f"  মোট PUBLISH চেষ্টা: {attempt}, সাবস্ক্রাইবার বার্তাটি মোট {len(deliveries)} বার পেলো "
                f"(ডুপ্লিকেট ঝুঁকি QoS1-এর স্বভাবজাত বৈশিষ্ট্য)")
    return log


def publish_qos2(broker, topic, payload):
    pid = broker.new_packet_id()
    log = [
        f"[QoS 2] ক্লায়েন্ট -> ব্রোকার: PUBLISH(id={pid})",
        f"  ব্রোকার -> ক্লায়েন্ট: PUBREC(id={pid})  (গ্রহণ করা হয়েছে, কিন্তু এখনও ডেলিভার হয়নি)",
        f"  ক্লায়েন্ট -> ব্রোকার: PUBREL(id={pid})  (রিলিজ নিশ্চিতকরণ)",
    ]
    recipients = broker.route(topic)
    log.append(f"  ব্রোকার এখন প্রকৃত ডেলিভারি করলো -> {recipients}")
    log.append(f"  ব্রোকার -> ক্লায়েন্ট: PUBCOMP(id={pid})  (হ্যান্ডশেক সম্পূর্ণ, ঠিক একবার ডেলিভারি নিশ্চিত)")
    return log


broker = MqttBroker()
broker.subscribe("dashboard-A", "sensors/+/temperature")
broker.subscribe("logger-B", "sensors/#")

print("== QoS 0 ==")
for line in publish_qos0(broker, "sensors/room1/temperature", 24.5):
    print(line)

print("\n== QoS 1 (স্বাভাবিক, কোনো লস নেই) ==")
for line in publish_qos1(broker, "sensors/room1/temperature", 24.6, simulate_lost_ack=False):
    print(line)

print("\n== QoS 1 (প্রথম PUBACK হারিয়ে গেছে -> জেনুইন রিট্রান্সমিশন) ==")
for line in publish_qos1(broker, "sensors/room1/temperature", 24.7, simulate_lost_ack=True):
    print(line)

print("\n== QoS 2 (৪-ধাপের হ্যান্ডশেক) ==")
for line in publish_qos2(broker, "sensors/room1/temperature", 24.8):
    print(line)

class WillAwareBroker(MqttBroker):
    """LWT সাপোর্টসহ ব্রোকার -- ক্লায়েন্ট প্রতি একটি ঐচ্ছিক (topic, payload) উইল রাখে।"""

    def __init__(self):
        super().__init__()
        self.wills = {}

    def connect(self, client_id, will_topic=None, will_payload=None):
        if will_topic is not None:
            self.wills[client_id] = (will_topic, will_payload)

    def disconnect_ungracefully(self, client_id):
        log = [f"ক্লায়েন্ট '{client_id}' অপ্রত্যাশিতভাবে সংযোগ হারালো (পরিষ্কার DISCONNECT পাঠায়নি)"]
        if client_id in self.wills:
            topic, payload = self.wills[client_id]
            recipients = self.route(topic)
            log.append(f"  ব্রোকার স্বয়ংক্রিয়ভাবে LWT প্রকাশ করলো -> টপিক='{topic}', payload='{payload}'")
            log.append(f"  ডেলিভার্ড: {recipients}")
        else:
            log.append(f"  ক্লায়েন্ট '{client_id}'-এর কোনো নিবন্ধিত উইল নেই -> কিছু প্রকাশ হলো না")
        return log


will_broker = WillAwareBroker()
will_broker.subscribe("status-monitor", "devices/+/status")
will_broker.connect("sensor-node-7", will_topic="devices/sensor-node-7/status", will_payload="offline")

print("সংযোগের সময় নিবন্ধিত উইল:", will_broker.wills)
print()
for line in will_broker.disconnect_ungracefully("sensor-node-7"):
    print(line)