"""Publish into a slow subscriber and see what the overflow policy buys.

    python3 demo.py

No threads, no sleeping, no randomness: integer ticks throughout.
"""

from broker import BLOCK, DROP_NEWEST, DROP_OLDEST, run

TICKS, PUBLISH_RATE, CAPACITY = 200, 5, 10
SLOW, FAST = 1, 5


def rule(title):
    print(f"\n{title}")
    print("-" * len(title))


print(f"One topic. Publisher emits {PUBLISH_RATE} messages/tick for {TICKS} ticks.")
print(f"Two subscribers: sub0 drains {SLOW}/tick (slow), sub1 drains {FAST}/tick.")
print(f"Every queue holds {CAPACITY}.")

# ---------------------------------------------------------------- 1
rule("1. The two drop policies, side by side")
print(f"{'policy':>12} {'published':>10} {'sub0 got':>9} {'sub0 lost':>10} {'loss':>7}")
runs = {}
for policy in (DROP_NEWEST, DROP_OLDEST):
    broker, subs = run(policy, TICKS, PUBLISH_RATE, CAPACITY, (SLOW, FAST))
    runs[policy] = (broker, subs)
    slow = subs[0]
    print(f"{policy:>12} {broker.published:>10} {len(slow.delivered):>9} "
          f"{slow.dropped:>10} {100 * slow.dropped / broker.published:>6.1f}%")

a = runs[DROP_NEWEST][1][0].dropped
b = runs[DROP_OLDEST][1][0].dropped
print(f"\nsame number lost either way: {a} vs {b} -> {'IDENTICAL' if a == b else 'DIFFERENT'}")

# ---------------------------------------------------------------- 2
rule("2. What the policy did change: which messages, and how stale")
for policy in (DROP_NEWEST, DROP_OLDEST):
    slow = runs[policy][1][0]
    seqs = [s for s, _, _ in slow.delivered]
    ages = slow.ages()
    print(f"{policy}:")
    print(f"    first 8 delivered: {seqs[:8]}")
    print(f"     last 8 delivered: {seqs[-8:]}")
    print(f"    mean age {sum(ages) / len(ages):.1f} ticks, final age {ages[-1]} ticks")
print("\nSame count, same queue occupancy, opposite ends of the stream.")

# ---------------------------------------------------------------- 3
rule("3. The loss is arithmetic, and the policy is not in it")


def predict(rate, drain, cap, ticks=TICKS):
    """Exact whenever the queue can hold one tick's burst (capacity >= rate)."""
    return max(0, (rate - drain) * ticks - (cap - drain))


print("    drops = (publish_rate - drain_rate) x ticks - (capacity - drain_rate)")
print(f"\n{'capacity':>9} {'drop-newest':>12} {'drop-oldest':>12} {'predicted':>10}")
for cap in (5, 10, 50, 100, 200, 500):
    n = run(DROP_NEWEST, TICKS, PUBLISH_RATE, cap, (SLOW,))[1][0].dropped
    o = run(DROP_OLDEST, TICKS, PUBLISH_RATE, cap, (SLOW,))[1][0].dropped
    print(f"{cap:>9} {n:>12} {o:>12} {predict(PUBLISH_RATE, SLOW, cap):>10}")
print("\nThe buffer is a one-time credit of capacity-drain_rate messages, not")
print(f"a rate. Going from 10 to 500 slots buys {500 - 10} messages, once,")
print(f"against a deficit of {PUBLISH_RATE - SLOW}/tick.")

print(f"\n{'publish rate':>13} {'drop-newest':>12} {'drop-oldest':>12} {'predicted':>10}")
for rate in (1, 2, 3, 5, 10):
    n = run(DROP_NEWEST, TICKS, rate, CAPACITY, (SLOW,))[1][0].dropped
    o = run(DROP_OLDEST, TICKS, rate, CAPACITY, (SLOW,))[1][0].dropped
    print(f"{rate:>13} {n:>12} {o:>12} {predict(rate, SLOW, CAPACITY):>10}")

print(f"\n{'drain rate':>13} {'drop-newest':>12} {'drop-oldest':>12} {'predicted':>10}")
for drain in (1, 2, 3, 4, 5):
    n = run(DROP_NEWEST, TICKS, PUBLISH_RATE, CAPACITY, (drain,))[1][0].dropped
    o = run(DROP_OLDEST, TICKS, PUBLISH_RATE, CAPACITY, (drain,))[1][0].dropped
    print(f"{drain:>13} {n:>12} {o:>12} {predict(PUBLISH_RATE, drain, CAPACITY):>10}")

# ---------------------------------------------------------------- 4
rule("4. Isolation: does the slow subscriber hurt the fast one?")
for policy in (DROP_NEWEST, DROP_OLDEST):
    broker, subs = runs[policy]
    print(f"{policy}:")
    for s in subs:
        print(f"    {s.name} (drain {s.drain_rate}/tick): "
              f"delivered {len(s.delivered):>4}, dropped {s.dropped:>4}")

# ---------------------------------------------------------------- 5
rule("5. Backpressure: the one policy that does change the number")
broker, subs = run(BLOCK, TICKS, PUBLISH_RATE, CAPACITY, (SLOW, FAST))
print(f"published {broker.published}, refused {broker.refused} "
      f"(publisher stalled on the slow subscriber)")
for s in subs:
    print(f"    {s.name} (drain {s.drain_rate}/tick): "
          f"delivered {len(s.delivered):>4}, dropped {s.dropped:>4}")
newest_fast = len(runs[DROP_NEWEST][1][1].delivered)
block_fast = len(subs[1].delivered)
print(f"\nNothing is dropped. Instead the healthy subscriber's delivery count")
print(f"falls from {newest_fast} to {block_fast} — the slow consumer's problem")
print(f"has become everybody's, which is what 'one slow subscriber doesn't")
print(f"block the others' stops being true the moment you choose this policy.")