"""
第 4 章 - 同步发送演示
=================================

confluent-kafka 没有原生的同步 send()，本文件演示
如何用 threading.Event + delivery callback 包装同步语义。

适用场景：
- 关键消息必须 ack 后才允许业务继续（订单写完才扣款）
- 单元测试需要确定性

性能：吞吐降到 1k-5k msg/s，比异步低 1-2 个数量级。

运行：
    bash ../init.sh
    python sync_producer.py             # 默认发 100 条
    python sync_producer.py 500 myhost:9092
"""

from __future__ import annotations

import sys
import threading
import time

from confluent_kafka import Producer

BOOTSTRAP = sys.argv[2] if len(sys.argv) > 2 else "127.0.0.1:9092"
N = int(sys.argv[1]) if len(sys.argv) > 1 else 100
TOPIC = "learn.04.demo"


class SyncSender:
    """把异步 Producer 包成「单条同步」。"""

    def __init__(self, producer: Producer):
        self._p = producer

    def send(self, topic: str, value: bytes, key: bytes | None = None, timeout: float = 10.0):
        done = threading.Event()
        result: dict = {}

        def cb(err, msg):
            result["err"] = err
            result["msg"] = msg
            done.set()

        self._p.produce(topic, key=key, value=value, on_delivery=cb)
        # 主动驱动：让 IO 线程把 batch 发出去
        self._p.poll(0)
        if not done.wait(timeout):
            raise TimeoutError(f"send timeout after {timeout}s")
        if result["err"] is not None:
            raise RuntimeError(f"delivery failed: {result['err']}")
        return result["msg"]


def main() -> None:
    producer = Producer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "client.id": "ch4-sync-producer",
            "acks": "all",
            "enable.idempotence": True,
            # 关键：同步语义下不要再攒批，linger=0 让消息立即发出
            "linger.ms": 0,
        }
    )
    sender = SyncSender(producer)

    print(f"🚀 同步发送 {N} 条到 {TOPIC}")
    start = time.perf_counter()
    for i in range(N):
        msg = sender.send(
            TOPIC,
            value=f"sync-msg-{i}".encode(),
            key=f"k{i % 5}".encode(),
        )
        if i < 5 or i == N - 1:
            print(
                f"  [{i:>4}] partition={msg.partition()} offset={msg.offset()} "
                f"latency_ms={(time.perf_counter() - start) * 1000 / (i + 1):.2f}"
            )

    elapsed = time.perf_counter() - start
    print(
        f"\n📊 完成 {N} 条；总耗时 {elapsed:.2f}s；"
        f"吞吐 {N / elapsed:.0f} msg/s；平均单条 {elapsed / N * 1000:.2f} ms"
    )
    producer.flush(5)


if __name__ == "__main__":
    main()