软件架构书籍路线：设计模式、系统思维与分布式设计

"""本站原创：本地进程崩溃与重投模型，不是真实消息代理或分布式事务。"""
import argparse
import os
import sqlite3


def apply_event(database, strategy, consumer, event_id, task_id):
    db = sqlite3.connect(database, isolation_level=None, timeout=5)
    try:
        db.executescript("""
            CREATE TABLE IF NOT EXISTS demo_counters (
                consumer TEXT PRIMARY KEY, value INTEGER NOT NULL);
            CREATE TABLE IF NOT EXISTS demo_receipts (
                consumer TEXT NOT NULL, event_id TEXT NOT NULL, task_id TEXT NOT NULL,
                PRIMARY KEY (consumer, event_id));
        """)
        db.execute("BEGIN IMMEDIATE")
        db.execute("INSERT OR IGNORE INTO demo_counters VALUES (?, 0)", (consumer,))
        previous = None
        if strategy == "dedup":
            previous = db.execute(
                "SELECT task_id FROM demo_receipts WHERE consumer=? AND event_id=?",
                (consumer, event_id)).fetchone()
        if previous is not None:
            if previous[0] != task_id:
                raise ValueError("same event ID carries a different task")
            outcome = "duplicate_suppressed"
        else:
            if strategy == "dedup":
                db.execute("INSERT INTO demo_receipts VALUES (?, ?, ?)",
                           (consumer, event_id, task_id))
            db.execute("UPDATE demo_counters SET value=value+1 WHERE consumer=?", (consumer,))
            outcome = "applied"
        db.execute("COMMIT")
        return outcome
    except Exception:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise
    finally:
        db.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("database")
    parser.add_argument("strategy", choices=["naive", "dedup"])
    parser.add_argument("consumer")
    parser.add_argument("event_id")
    parser.add_argument("task_id")
    parser.add_argument("--crash-after-commit", action="store_true")
    args = parser.parse_args()
    result = apply_event(args.database, args.strategy, args.consumer, args.event_id, args.task_id)
    print(result + ": committed", flush=True)
    if args.crash_after_commit:
        os._exit(70)  # Actual child-process exit before the modeled acknowledgement.
    print("ack", flush=True)  # A local marker, not a real broker acknowledgement.