# history

source:
  import subprocess, time, json

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      """Arranca un broker NATS en Docker de forma idempotente. Devuelve el estado."""
      # ¿Existe el contenedor (corriendo o parado)?
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name)
          state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222")
          state = "created"
      # Esperar a que el puerto acepte conexiones
      time.sleep(1.0)
      return state

  state = ensure_nats()
  print(f"Broker NATS: {state} en {NATS_URL}")
  # El endpoint de monitorización (puerto 8222) expone métricas del servidor en JSON
  import urllib.request
  varz = json.loads(urllib.request.urlopen("http://127.0.0.1:8222/varz", timeout=3).read())
  print(f"nats-server version: {varz['version']} | jetstream activo: {'jetstream' in varz and bool(varz['jetstream'])}")

source_raw:
  import subprocess, time, json

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      """Arranca un broker NATS en Docker de forma idempotente. Devuelve el estado."""
      # ¿Existe el contenedor (corriendo o parado)?
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name)
          state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222")
          state = "created"
      # Esperar a que el puerto acepte conexiones
      time.sleep(1.0)
      return state

  state = ensure_nats()
  print(f"Broker NATS: {state} en {NATS_URL}")
  # El endpoint de monitorización (puerto 8222) expone métricas del servidor en JSON
  import urllib.request
  varz = json.loads(urllib.request.urlopen("http://127.0.0.1:8222/varz", timeout=3).read())
  print(f"nats-server version: {varz['version']} | jetstream activo: {'jetstream' in varz and bool(varz['jetstream'])}")


source:
  import asyncio
  import nats

  # Conexión persistente del notebook (simula el cliente de un proceso de larga vida)
  nc = await nats.connect(NATS_URL, name="notebook-01")
  info = nc._server_info  # metadata que el broker envía en el handshake
  print("Conectado.")
  print(f"  server_id  : {info['server_id']}")
  print(f"  max_payload: {info['max_payload']/1024/1024:.0f} MiB por mensaje")
  print(f"  client_id  : {info['client_id']}")

source_raw:
  import asyncio
  import nats

  # Conexión persistente del notebook (simula el cliente de un proceso de larga vida)
  nc = await nats.connect(NATS_URL, name="notebook-01")
  info = nc._server_info  # metadata que el broker envía en el handshake
  print("Conectado.")
  print(f"  server_id  : {info['server_id']}")
  print(f"  max_payload: {info['max_payload']/1024/1024:.0f} MiB por mensaje")
  print(f"  client_id  : {info['client_id']}")


source:
  # El subscriber declara interés ANTES de que se publique (core = at-most-once)
  sub = await nc.subscribe("saludos")

  # El publisher envía. No sabe quién escucha; solo conoce el subject.
  await nc.publish("saludos", b"hola desde el publisher")
  await nc.flush()  # fuerza el envío al broker antes de seguir

  # El subscriber recoge el mensaje
  msg = await sub.next_msg(timeout=2)
  print(f"subject recibido : {msg.subject}")
  print(f"payload          : {msg.data.decode()}")

  await sub.unsubscribe()

source_raw:
  # El subscriber declara interés ANTES de que se publique (core = at-most-once)
  sub = await nc.subscribe("saludos")

  # El publisher envía. No sabe quién escucha; solo conoce el subject.
  await nc.publish("saludos", b"hola desde el publisher")
  await nc.flush()  # fuerza el envío al broker antes de seguir

  # El subscriber recoge el mensaje
  msg = await sub.next_msg(timeout=2)
  print(f"subject recibido : {msg.subject}")
  print(f"payload          : {msg.data.decode()}")

  await sub.unsubscribe()


source:
  recibidos = []  # registro de quién recibió qué

  def make_handler(nombre):
      async def handler(msg):
          recibidos.append({"subscriber": nombre, "subject": msg.subject, "data": msg.data.decode()})
      return handler

  # Tres subscribers independientes al MISMO subject
  subs = []
  for nombre in ["sub-A", "sub-B", "sub-C"]:
      s = await nc.subscribe("noticias", cb=make_handler(nombre))
      subs.append(s)

  # Un único publish...
  await nc.publish("noticias", b"titular: NATS entrega a todos")
  await nc.flush()
  await asyncio.sleep(0.3)  # dar tiempo a los callbacks

  for r in recibidos:
      print(f"{r['subscriber']} <- [{r['subject']}] {r['data']}")
  print(f"\nUn publish -> {len(recibidos)} entregas (fan-out)")

  for s in subs:
      await s.unsubscribe()

source_raw:
  recibidos = []  # registro de quién recibió qué

  def make_handler(nombre):
      async def handler(msg):
          recibidos.append({"subscriber": nombre, "subject": msg.subject, "data": msg.data.decode()})
      return handler

  # Tres subscribers independientes al MISMO subject
  subs = []
  for nombre in ["sub-A", "sub-B", "sub-C"]:
      s = await nc.subscribe("noticias", cb=make_handler(nombre))
      subs.append(s)

  # Un único publish...
  await nc.publish("noticias", b"titular: NATS entrega a todos")
  await nc.flush()
  await asyncio.sleep(0.3)  # dar tiempo a los callbacks

  for r in recibidos:
      print(f"{r['subscriber']} <- [{r['subject']}] {r['data']}")
  print(f"\nUn publish -> {len(recibidos)} entregas (fan-out)")

  for s in subs:
      await s.unsubscribe()


source:
  wild = []

  async def on_star(msg):
      wild.append({"patron": "sensor.*.temp", "subject": msg.subject, "data": msg.data.decode()})

  async def on_gt(msg):
      wild.append({"patron": "sensor.>", "subject": msg.subject, "data": msg.data.decode()})

  s_star = await nc.subscribe("sensor.*.temp", cb=on_star)
  s_gt   = await nc.subscribe("sensor.>",      cb=on_gt)

  # Publicamos en varios subjects concretos
  publicados = {
      "sensor.cocina.temp":          "21.5",
      "sensor.salon.temp":           "22.1",
      "sensor.salon.humedad":        "48",
      "sensor.garaje.puerta.estado": "abierta",
  }
  for subj, val in publicados.items():
      await nc.publish(subj, val.encode())
  await nc.flush()
  await asyncio.sleep(0.3)

  print("Mensajes publicados:", list(publicados))
  print()
  for w in wild:
      print(f"[{w['patron']:14}] casó {w['subject']}")

  await s_star.unsubscribe(); await s_gt.unsubscribe()

source_raw:
  wild = []

  async def on_star(msg):
      wild.append({"patron": "sensor.*.temp", "subject": msg.subject, "data": msg.data.decode()})

  async def on_gt(msg):
      wild.append({"patron": "sensor.>", "subject": msg.subject, "data": msg.data.decode()})

  s_star = await nc.subscribe("sensor.*.temp", cb=on_star)
  s_gt   = await nc.subscribe("sensor.>",      cb=on_gt)

  # Publicamos en varios subjects concretos
  publicados = {
      "sensor.cocina.temp":          "21.5",
      "sensor.salon.temp":           "22.1",
      "sensor.salon.humedad":        "48",
      "sensor.garaje.puerta.estado": "abierta",
  }
  for subj, val in publicados.items():
      await nc.publish(subj, val.encode())
  await nc.flush()
  await asyncio.sleep(0.3)

  print("Mensajes publicados:", list(publicados))
  print()
  for w in wild:
      print(f"[{w['patron']:14}] casó {w['subject']}")

  await s_star.unsubscribe(); await s_gt.unsubscribe()


source:
  import pandas as pd
  import matplotlib.pyplot as plt
  import numpy as np

  patrones = ["sensor.*.temp", "sensor.>"]
  subjects = list(publicados)

  # Construir matriz de coincidencias a partir de lo que recibió cada suscripción
  M = np.zeros((len(patrones), len(subjects)), dtype=int)
  for w in wild:
      i = patrones.index(w["patron"])
      j = subjects.index(w["subject"])
      M[i, j] = 1

  fig, ax = plt.subplots(figsize=(9, 2.6))
  ax.imshow(M, cmap="Greens", vmin=0, vmax=1, aspect="auto")
  ax.set_xticks(range(len(subjects))); ax.set_xticklabels(subjects, rotation=25, ha="right", fontsize=9)
  ax.set_yticks(range(len(patrones))); ax.set_yticklabels(patrones, fontsize=10)
  for i in range(len(patrones)):
      for j in range(len(subjects)):
          ax.text(j, i, "OK" if M[i, j] else "-", ha="center", va="center",
                  color="white" if M[i, j] else "#999", fontsize=12)
  ax.set_title("Coincidencia de wildcards (OK = el subscriber recibió el mensaje)")
  plt.tight_layout(); plt.show()

  pd.DataFrame(M, index=patrones, columns=subjects)

source_raw:
  import pandas as pd
  import matplotlib.pyplot as plt
  import numpy as np

  patrones = ["sensor.*.temp", "sensor.>"]
  subjects = list(publicados)

  # Construir matriz de coincidencias a partir de lo que recibió cada suscripción
  M = np.zeros((len(patrones), len(subjects)), dtype=int)
  for w in wild:
      i = patrones.index(w["patron"])
      j = subjects.index(w["subject"])
      M[i, j] = 1

  fig, ax = plt.subplots(figsize=(9, 2.6))
  ax.imshow(M, cmap="Greens", vmin=0, vmax=1, aspect="auto")
  ax.set_xticks(range(len(subjects))); ax.set_xticklabels(subjects, rotation=25, ha="right", fontsize=9)
  ax.set_yticks(range(len(patrones))); ax.set_yticklabels(patrones, fontsize=10)
  for i in range(len(patrones)):
      for j in range(len(subjects)):
          ax.text(j, i, "OK" if M[i, j] else "-", ha="center", va="center",
                  color="white" if M[i, j] else "#999", fontsize=12)
  ax.set_title("Coincidencia de wildcards (OK = el subscriber recibió el mensaje)")
  plt.tight_layout(); plt.show()

  pd.DataFrame(M, index=patrones, columns=subjects)


source:
  import subprocess, time, asyncio
  import nats

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name); state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222"); state = "created"
      time.sleep(1.0)
      return state

  print("Broker:", ensure_nats())
  nc = await nats.connect(NATS_URL, name="notebook-02")
  print("Conectado, client_id:", nc._server_info["client_id"])

source_raw:
  import subprocess, time, asyncio
  import nats

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name); state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222"); state = "created"
      time.sleep(1.0)
      return state

  print("Broker:", ensure_nats())
  nc = await nats.connect(NATS_URL, name="notebook-02")
  print("Conectado, client_id:", nc._server_info["client_id"])


source:
  from collections import Counter

  trabajo = Counter()      # cuántas tareas procesó cada worker
  orden = []               # traza temporal (worker, tarea)

  def make_worker(nombre):
      async def worker(msg):
          tarea = msg.data.decode()
          trabajo[nombre] += 1
          orden.append((nombre, tarea))
      return worker

  # 3 workers, MISMO subject, MISMO queue group -> NATS reparte
  workers = []
  for nombre in ["worker-1", "worker-2", "worker-3"]:
      s = await nc.subscribe("tareas", queue="procesadores", cb=make_worker(nombre))
      workers.append(s)

  # Publicar 12 tareas
  for i in range(12):
      await nc.publish("tareas", f"tarea-{i:02d}".encode())
  await nc.flush()
  await asyncio.sleep(0.5)

  print("Reparto de carga (cada tarea fue a UN solo worker):")
  for w, n in sorted(trabajo.items()):
      print(f"  {w}: {n} tareas")
  print(f"  TOTAL procesado: {sum(trabajo.values())} de 12 tareas")

  for s in workers:
      await s.unsubscribe()

source_raw:
  from collections import Counter

  trabajo = Counter()      # cuántas tareas procesó cada worker
  orden = []               # traza temporal (worker, tarea)

  def make_worker(nombre):
      async def worker(msg):
          tarea = msg.data.decode()
          trabajo[nombre] += 1
          orden.append((nombre, tarea))
      return worker

  # 3 workers, MISMO subject, MISMO queue group -> NATS reparte
  workers = []
  for nombre in ["worker-1", "worker-2", "worker-3"]:
      s = await nc.subscribe("tareas", queue="procesadores", cb=make_worker(nombre))
      workers.append(s)

  # Publicar 12 tareas
  for i in range(12):
      await nc.publish("tareas", f"tarea-{i:02d}".encode())
  await nc.flush()
  await asyncio.sleep(0.5)

  print("Reparto de carga (cada tarea fue a UN solo worker):")
  for w, n in sorted(trabajo.items()):
      print(f"  {w}: {n} tareas")
  print(f"  TOTAL procesado: {sum(trabajo.values())} de 12 tareas")

  for s in workers:
      await s.unsubscribe()


source:
  import matplotlib.pyplot as plt

  nombres = [f"worker-{i}" for i in (1, 2, 3)]
  valores = [trabajo.get(n, 0) for n in nombres]

  fig, ax = plt.subplots(figsize=(7, 3))
  barras = ax.bar(nombres, valores, color=["#2563eb", "#16a34a", "#db2777"])
  ax.bar_label(barras, padding=3)
  ax.set_ylabel("tareas procesadas")
  ax.set_title(f"Queue group 'procesadores' — 12 tareas repartidas entre {len(nombres)} workers")
  ax.set_ylim(0, max(valores) + 2)
  plt.tight_layout(); plt.show()

source_raw:
  import matplotlib.pyplot as plt

  nombres = [f"worker-{i}" for i in (1, 2, 3)]
  valores = [trabajo.get(n, 0) for n in nombres]

  fig, ax = plt.subplots(figsize=(7, 3))
  barras = ax.bar(nombres, valores, color=["#2563eb", "#16a34a", "#db2777"])
  ax.bar_label(barras, padding=3)
  ax.set_ylabel("tareas procesadas")
  ax.set_title(f"Queue group 'procesadores' — 12 tareas repartidas entre {len(nombres)} workers")
  ax.set_ylim(0, max(valores) + 2)
  plt.tight_layout(); plt.show()


source:
  # Servicio: convierte el texto recibido a mayúsculas y responde
  async def servicio_mayusculas(msg):
      respuesta = msg.data.decode().upper()
      await msg.respond(respuesta.encode())

  sub_svc = await nc.subscribe("servicio.mayusculas", cb=servicio_mayusculas)

  # Cliente: pide y espera respuesta (con timeout)
  peticiones = ["hola mundo", "nats request reply", "desacople total"]
  for texto in peticiones:
      resp = await nc.request("servicio.mayusculas", texto.encode(), timeout=1.0)
      print(f"  peticion : {texto!r}")
      print(f"  respuesta: {resp.data.decode()!r}  (vino por el inbox {resp.subject})")
      print()

  # Timeout cuando nadie responde a un subject
  try:
      await nc.request("servicio.inexistente", b"hay alguien?", timeout=0.5)
  except asyncio.TimeoutError:
      print("servicio.inexistente -> TimeoutError: nadie respondió (comportamiento esperado)")

  await sub_svc.unsubscribe()

source_raw:
  # Servicio: convierte el texto recibido a mayúsculas y responde
  async def servicio_mayusculas(msg):
      respuesta = msg.data.decode().upper()
      await msg.respond(respuesta.encode())

  sub_svc = await nc.subscribe("servicio.mayusculas", cb=servicio_mayusculas)

  # Cliente: pide y espera respuesta (con timeout)
  peticiones = ["hola mundo", "nats request reply", "desacople total"]
  for texto in peticiones:
      resp = await nc.request("servicio.mayusculas", texto.encode(), timeout=1.0)
      print(f"  peticion : {texto!r}")
      print(f"  respuesta: {resp.data.decode()!r}  (vino por el inbox {resp.subject})")
      print()

  # Timeout cuando nadie responde a un subject
  try:
      await nc.request("servicio.inexistente", b"hay alguien?", timeout=0.5)
  except asyncio.TimeoutError:
      print("servicio.inexistente -> TimeoutError: nadie respondió (comportamiento esperado)")

  await sub_svc.unsubscribe()


source:
  js = nc.jetstream()

  # Crear (o actualizar) un stream que persiste todo lo de 'eventos.>'
  # Lo recreamos limpio para que la demo sea determinista
  try:
      await js.delete_stream("EVENTOS")
  except Exception:
      pass
  stream = await js.add_stream(name="EVENTOS", subjects=["eventos.>"])
  print(f"Stream creado: {stream.config.name}  subjects={stream.config.subjects}  storage={stream.config.storage}")

  # Publicar 5 eventos SIN que haya ningún consumidor escuchando todavía
  for i in range(5):
      ack = await js.publish("eventos.click", f"evento #{i}".encode())
      print(f"  publicado eventos.click -> stream='{ack.stream}' seq={ack.seq}")

  info = await js.find_stream_name_by_subject("eventos.click")
  estado = (await js.stream_info("EVENTOS")).state
  print(f"\nMensajes retenidos en el stream: {estado.messages} (siguen ahí aunque nadie los haya leído)")

source_raw:
  js = nc.jetstream()

  # Crear (o actualizar) un stream que persiste todo lo de 'eventos.>'
  # Lo recreamos limpio para que la demo sea determinista
  try:
      await js.delete_stream("EVENTOS")
  except Exception:
      pass
  stream = await js.add_stream(name="EVENTOS", subjects=["eventos.>"])
  print(f"Stream creado: {stream.config.name}  subjects={stream.config.subjects}  storage={stream.config.storage}")

  # Publicar 5 eventos SIN que haya ningún consumidor escuchando todavía
  for i in range(5):
      ack = await js.publish("eventos.click", f"evento #{i}".encode())
      print(f"  publicado eventos.click -> stream='{ack.stream}' seq={ack.seq}")

  info = await js.find_stream_name_by_subject("eventos.click")
  estado = (await js.stream_info("EVENTOS")).state
  print(f"\nMensajes retenidos en el stream: {estado.messages} (siguen ahí aunque nadie los haya leído)")


source:
  # Pull consumer durable: pedimos los mensajes bajo demanda
  psub = await js.pull_subscribe("eventos.>", durable="lector-eventos")

  mensajes = await psub.fetch(5, timeout=2)
  print(f"Recuperados {len(mensajes)} mensajes del stream (replay):")
  for m in mensajes:
      print(f"  seq={m.metadata.sequence.stream}  subject={m.subject}  data={m.data.decode()!r}")
      await m.ack()  # confirmamos el procesado; el durable avanza su cursor

  # Tras el ack, un segundo fetch no devuelve nada nuevo (cursor avanzado)
  try:
      extra = await psub.fetch(1, timeout=1)
  except (asyncio.TimeoutError, Exception):
      extra = []
  print(f"\nSegundo fetch tras ack: {len(extra)} mensajes (el durable recuerda que ya los leyó)")

source_raw:
  # Pull consumer durable: pedimos los mensajes bajo demanda
  psub = await js.pull_subscribe("eventos.>", durable="lector-eventos")

  mensajes = await psub.fetch(5, timeout=2)
  print(f"Recuperados {len(mensajes)} mensajes del stream (replay):")
  for m in mensajes:
      print(f"  seq={m.metadata.sequence.stream}  subject={m.subject}  data={m.data.decode()!r}")
      await m.ack()  # confirmamos el procesado; el durable avanza su cursor

  # Tras el ack, un segundo fetch no devuelve nada nuevo (cursor avanzado)
  try:
      extra = await psub.fetch(1, timeout=1)
  except (asyncio.TimeoutError, Exception):
      extra = []
  print(f"\nSegundo fetch tras ack: {len(extra)} mensajes (el durable recuerda que ya los leyó)")


source:
  import subprocess, time, json

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      """Arranca un broker NATS en Docker de forma idempotente. Devuelve el estado."""
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name)
          state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222")
          state = "created"
      time.sleep(1.0)
      return state

  import asyncio
  import nats

  print("Broker:", ensure_nats())
  nc = await nats.connect(NATS_URL, name="notebook-04")
  js = nc.jetstream()
  print("JetStream context listo. account info:")
  ai = await js.account_info()
  print(f"  streams={ai.streams}  consumers={ai.consumers}  memory={ai.memory}  storage={ai.storage}")

source_raw:
  import subprocess, time, json

  NATS_CONTAINER = "nats_demo"
  NATS_PORT = 4222
  NATS_URL = f"nats://127.0.0.1:{NATS_PORT}"

  def _docker(*args, check=True):
      return subprocess.run(["docker", *args], capture_output=True, text=True, check=check)

  def ensure_nats(name=NATS_CONTAINER, port=NATS_PORT):
      """Arranca un broker NATS en Docker de forma idempotente. Devuelve el estado."""
      out = _docker("ps", "-a", "--filter", f"name=^{name}$", "--format", "{{.State}}", check=False).stdout.strip()
      if out == "running":
          state = "already-running"
      elif out in ("exited", "created", "paused"):
          _docker("start", name)
          state = "started"
      else:
          _docker("run", "-d", "--name", name, "-p", f"{port}:4222", "-p", "8222:8222",
                  "nats:latest", "-js", "-m", "8222")
          state = "created"
      time.sleep(1.0)
      return state

  import asyncio
  import nats

  print("Broker:", ensure_nats())
  nc = await nats.connect(NATS_URL, name="notebook-04")
  js = nc.jetstream()
  print("JetStream context listo. account info:")
  ai = await js.account_info()
  print(f"  streams={ai.streams}  consumers={ai.consumers}  memory={ai.memory}  storage={ai.storage}")


source:
  from nats.js.api import StreamConfig, RetentionPolicy, StorageType, DiscardPolicy

  # Recrear limpio para que la demo sea determinista
  for s in ("DEMO_LIMITS", "DEMO_DEDUP", "DEMO_WQ"):
      try:
          await js.delete_stream(s)
      except Exception:
          pass

  cfg = StreamConfig(
      name="DEMO_LIMITS",
      subjects=["demo.limits.>"],
      storage=StorageType.FILE,
      retention=RetentionPolicy.LIMITS,
      max_msgs=1000,                 # tope de mensajes
      max_age=3600,                  # 1 hora (segundos)
      discard=DiscardPolicy.OLD,     # al llegar al tope, descarta los más viejos
      duplicate_window=120,          # ventana de dedup: 120 s
  )
  info = await js.add_stream(cfg)
  c = info.config
  print("Stream creado:")
  print(f"  name          : {c.name}")
  print(f"  subjects      : {c.subjects}")
  print(f"  storage       : {c.storage}")
  print(f"  retention     : {c.retention}")
  print(f"  max_msgs      : {c.max_msgs}")
  print(f"  max_age (s)   : {c.max_age}")
  print(f"  discard       : {c.discard}")
  print(f"  dup_window (s): {c.duplicate_window}")

source_raw:
  from nats.js.api import StreamConfig, RetentionPolicy, StorageType, DiscardPolicy

  # Recrear limpio para que la demo sea determinista
  for s in ("DEMO_LIMITS", "DEMO_DEDUP", "DEMO_WQ"):
      try:
          await js.delete_stream(s)
      except Exception:
          pass

  cfg = StreamConfig(
      name="DEMO_LIMITS",
      subjects=["demo.limits.>"],
      storage=StorageType.FILE,
      retention=RetentionPolicy.LIMITS,
      max_msgs=1000,                 # tope de mensajes
      max_age=3600,                  # 1 hora (segundos)
      discard=DiscardPolicy.OLD,     # al llegar al tope, descarta los más viejos
      duplicate_window=120,          # ventana de dedup: 120 s
  )
  info = await js.add_stream(cfg)
  c = info.config
  print("Stream creado:")
  print(f"  name          : {c.name}")
  print(f"  subjects      : {c.subjects}")
  print(f"  storage       : {c.storage}")
  print(f"  retention     : {c.retention}")
  print(f"  max_msgs      : {c.max_msgs}")
  print(f"  max_age (s)   : {c.max_age}")
  print(f"  discard       : {c.discard}")
  print(f"  dup_window (s): {c.duplicate_window}")


source:
  # Publicar 6 mensajes en el stream de límites
  for i in range(6):
      await js.publish("demo.limits.eventos", f"evento-{i}".encode())

  # Pull consumer DURABLE: recuerda su cursor entre fetches
  psub = await js.pull_subscribe("demo.limits.>", durable="procesador-A")

  # Traer 4 y confirmarlos (ack)
  batch = await psub.fetch(4, timeout=2)
  print("Primer fetch (4 msgs):")
  for m in batch:
      print(f"  seq={m.metadata.sequence.stream}  {m.data.decode()}")
      await m.ack()

  # Estado del consumer: cuántos quedan pendientes / entregados
  ci = await psub.consumer_info()
  print()
  print(f"Consumer 'procesador-A':")
  print(f"  num_pending    : {ci.num_pending}      (mensajes sin entregar todavía)")
  print(f"  num_ack_pending: {ci.num_ack_pending}  (entregados sin ack)")
  print(f"  delivered.stream_seq: {ci.delivered.stream_seq}")

  # Segundo fetch: continúa donde se quedó (recuerda el cursor)
  batch2 = await psub.fetch(10, timeout=1)
  print()
  print(f"Segundo fetch: {len(batch2)} msgs restantes ->", [m.data.decode() for m in batch2])
  for m in batch2:
      await m.ack()

source_raw:
  # Publicar 6 mensajes en el stream de límites
  for i in range(6):
      await js.publish("demo.limits.eventos", f"evento-{i}".encode())

  # Pull consumer DURABLE: recuerda su cursor entre fetches
  psub = await js.pull_subscribe("demo.limits.>", durable="procesador-A")

  # Traer 4 y confirmarlos (ack)
  batch = await psub.fetch(4, timeout=2)
  print("Primer fetch (4 msgs):")
  for m in batch:
      print(f"  seq={m.metadata.sequence.stream}  {m.data.decode()}")
      await m.ack()

  # Estado del consumer: cuántos quedan pendientes / entregados
  ci = await psub.consumer_info()
  print()
  print(f"Consumer 'procesador-A':")
  print(f"  num_pending    : {ci.num_pending}      (mensajes sin entregar todavía)")
  print(f"  num_ack_pending: {ci.num_ack_pending}  (entregados sin ack)")
  print(f"  delivered.stream_seq: {ci.delivered.stream_seq}")

  # Segundo fetch: continúa donde se quedó (recuerda el cursor)
  batch2 = await psub.fetch(10, timeout=1)
  print()
  print(f"Segundo fetch: {len(batch2)} msgs restantes ->", [m.data.decode() for m in batch2])
  for m in batch2:
      await m.ack()


source:
  await js.add_stream(name="DEMO_DEDUP", subjects=["demo.dedup.>"],
                      storage=StorageType.FILE, duplicate_window=120)

  # Publicar dos veces el MISMO Nats-Msg-Id
  ack1 = await js.publish("demo.dedup.pago", b"cobro 50e", headers={"Nats-Msg-Id": "pago-0001"})
  ack2 = await js.publish("demo.dedup.pago", b"cobro 50e", headers={"Nats-Msg-Id": "pago-0001"})

  print(f"1a publicacion: seq={ack1.seq}  duplicate={ack1.duplicate}")
  print(f"2a publicacion: seq={ack2.seq}  duplicate={ack2.duplicate}  <- detectada como duplicado")

  # Un Msg-Id distinto sí se almacena
  ack3 = await js.publish("demo.dedup.pago", b"cobro 30e", headers={"Nats-Msg-Id": "pago-0002"})
  print(f"3a publicacion (id nuevo): seq={ack3.seq}  duplicate={ack3.duplicate}")

  st = (await js.stream_info("DEMO_DEDUP")).state
  print()
  print(f"Mensajes realmente almacenados en el stream: {st.messages} (2 publicaciones unicas, 1 descartada)")

source_raw:
  await js.add_stream(name="DEMO_DEDUP", subjects=["demo.dedup.>"],
                      storage=StorageType.FILE, duplicate_window=120)

  # Publicar dos veces el MISMO Nats-Msg-Id
  ack1 = await js.publish("demo.dedup.pago", b"cobro 50e", headers={"Nats-Msg-Id": "pago-0001"})
  ack2 = await js.publish("demo.dedup.pago", b"cobro 50e", headers={"Nats-Msg-Id": "pago-0001"})

  print(f"1a publicacion: seq={ack1.seq}  duplicate={ack1.duplicate}")
  print(f"2a publicacion: seq={ack2.seq}  duplicate={ack2.duplicate}  <- detectada como duplicado")

  # Un Msg-Id distinto sí se almacena
  ack3 = await js.publish("demo.dedup.pago", b"cobro 30e", headers={"Nats-Msg-Id": "pago-0002"})
  print(f"3a publicacion (id nuevo): seq={ack3.seq}  duplicate={ack3.duplicate}")

  st = (await js.stream_info("DEMO_DEDUP")).state
  print()
  print(f"Mensajes realmente almacenados en el stream: {st.messages} (2 publicaciones unicas, 1 descartada)")


source:
  from nats.js.api import RetentionPolicy, StorageType

  await js.add_stream(name="DEMO_WQ", subjects=["demo.wq.>"],
                      storage=StorageType.FILE, retention=RetentionPolicy.WORK_QUEUE)

  # Encolar 5 tareas
  for i in range(5):
      await js.publish("demo.wq.tareas", f"tarea-{i}".encode())

  antes = (await js.stream_info("DEMO_WQ")).state.messages
  print(f"Tareas encoladas en el stream: {antes}")

  # Un worker consume y confirma 3
  wsub = await js.pull_subscribe("demo.wq.>", durable="worker")
  tres = await wsub.fetch(3, timeout=2)
  for m in tres:
      print(f"  procesada: {m.data.decode()}")
      await m.ack()   # al confirmar, JetStream BORRA el mensaje del stream

  await asyncio.sleep(0.3)
  despues = (await js.stream_info("DEMO_WQ")).state.messages
  print()
  print(f"Mensajes restantes en el stream tras 3 acks: {despues} (workqueue borra lo confirmado: {antes} -> {despues})")

source_raw:
  from nats.js.api import RetentionPolicy, StorageType

  await js.add_stream(name="DEMO_WQ", subjects=["demo.wq.>"],
                      storage=StorageType.FILE, retention=RetentionPolicy.WORK_QUEUE)

  # Encolar 5 tareas
  for i in range(5):
      await js.publish("demo.wq.tareas", f"tarea-{i}".encode())

  antes = (await js.stream_info("DEMO_WQ")).state.messages
  print(f"Tareas encoladas en el stream: {antes}")

  # Un worker consume y confirma 3
  wsub = await js.pull_subscribe("demo.wq.>", durable="worker")
  tres = await wsub.fetch(3, timeout=2)
  for m in tres:
      print(f"  procesada: {m.data.decode()}")
      await m.ack()   # al confirmar, JetStream BORRA el mensaje del stream

  await asyncio.sleep(0.3)
  despues = (await js.stream_info("DEMO_WQ")).state.messages
  print()
  print(f"Mensajes restantes en el stream tras 3 acks: {despues} (workqueue borra lo confirmado: {antes} -> {despues})")


source:
  from nats.js.api import ConsumerConfig, DeliverPolicy

  # Consumer que reproduce TODO el historial
  all_sub = await js.pull_subscribe(
      "demo.limits.>", durable="replay-all",
      config=ConsumerConfig(deliver_policy=DeliverPolicy.ALL),
  )
  todos = await all_sub.fetch(50, timeout=1)
  print(f"DeliverPolicy.ALL  -> {len(todos)} mensajes:", [m.data.decode() for m in todos])
  for m in todos:
      await m.ack()

  # Consumer que solo entrega el ÚLTIMO
  last_sub = await js.pull_subscribe(
      "demo.limits.>", durable="replay-last",
      config=ConsumerConfig(deliver_policy=DeliverPolicy.LAST),
  )
  ultimo = await last_sub.fetch(50, timeout=1)
  print(f"DeliverPolicy.LAST -> {len(ultimo)} mensaje :", [m.data.decode() for m in ultimo])
  for m in ultimo:
      await m.ack()

source_raw:
  from nats.js.api import ConsumerConfig, DeliverPolicy

  # Consumer que reproduce TODO el historial
  all_sub = await js.pull_subscribe(
      "demo.limits.>", durable="replay-all",
      config=ConsumerConfig(deliver_policy=DeliverPolicy.ALL),
  )
  todos = await all_sub.fetch(50, timeout=1)
  print(f"DeliverPolicy.ALL  -> {len(todos)} mensajes:", [m.data.decode() for m in todos])
  for m in todos:
      await m.ack()

  # Consumer que solo entrega el ÚLTIMO
  last_sub = await js.pull_subscribe(
      "demo.limits.>", durable="replay-last",
      config=ConsumerConfig(deliver_policy=DeliverPolicy.LAST),
  )
  ultimo = await last_sub.fetch(50, timeout=1)
  print(f"DeliverPolicy.LAST -> {len(ultimo)} mensaje :", [m.data.decode() for m in ultimo])
  for m in ultimo:
      await m.ack()


source:
  import ipywidgets as widgets
  from IPython.display import display, clear_output
  import matplotlib.pyplot as plt
  import asyncio, time
  import nats

  # --- widgets ---
  n_msgs_w = widgets.IntSlider(value=20000, min=1000, max=100000, step=1000,
                               description="Mensajes:", style={"description_width": "initial"},
                               layout=widgets.Layout(width="380px"))
  n_subs_w = widgets.IntSlider(value=3, min=1, max=8, step=1,
                               description="Subscribers:", style={"description_width": "initial"})
  run_btn  = widgets.Button(description="▶ Ejecutar benchmark", button_style="success",
                            layout=widgets.Layout(width="220px"))
  plot_out = widgets.Output()
  log_out  = widgets.Output()

  SUBJECT = "bench.load"
  PAYLOAD = b"x" * 128   # 128 bytes por mensaje

  def _throughput(ts, recv):
      thr = [0.0]
      for i in range(1, len(ts)):
          dt = ts[i] - ts[i-1]
          thr.append((recv[i] - recv[i-1]) / dt if dt > 0 else 0.0)
      return thr

  def render(history, n_subs, n_msgs, done=False):
      ts   = [h[0] for h in history]
      sent = [h[1] for h in history]
      recv = [h[2] for h in history]
      thr  = _throughput(ts, recv)
      with plot_out:
          clear_output(wait=True)
          fig, (a1, a2) = plt.subplots(1, 2, figsize=(11, 3.6))
          a1.plot(ts, sent, label="enviados (pub)", color="#2563eb", lw=2)
          a1.plot(ts, recv, label=f"recibidos (Σ {n_subs} subs)", color="#16a34a", lw=2)
          a1.set_xlabel("segundos"); a1.set_ylabel("mensajes acumulados")
          a1.set_title("Publisher vs subscribers"); a1.legend(loc="upper left")
          a2.plot(ts, thr, color="#db2777", lw=2)
          a2.set_xlabel("segundos"); a2.set_ylabel("msgs/s recibidos")
          a2.set_title("Throughput instantáneo")
          estado = "✓ DONE" if done else "● corriendo…"
          fig.suptitle(f"[{estado}]  {n_msgs:,} msgs → {n_subs} subs   "
                       f"enviados={sent[-1]:,}  recibidos={recv[-1]:,}", fontsize=11)
          plt.tight_layout(); plt.show()

  async def run_benchmark(n_msgs, n_subs, live=True):
      """1 publisher -> n_subs subscribers. Devuelve (history, counters)."""
      nc = await nats.connect(NATS_URL, name="benchmark")
      counters = [0] * n_subs

      def make_cb(i):
          async def cb(msg):
              counters[i] += 1
          return cb

      subs = [await nc.subscribe(SUBJECT, cb=make_cb(i)) for i in range(n_subs)]
      history = []          # (t, enviados, recibidos_total)
      sent = 0
      t0 = time.monotonic()

      async def publish_all():
          nonlocal sent
          for k in range(n_msgs):
              await nc.publish(SUBJECT, PAYLOAD)
              sent += 1
              if k % 1000 == 0:
                  await nc.flush()
                  await asyncio.sleep(0)   # ceder al event loop (deja correr callbacks)
          await nc.flush()

      task = asyncio.create_task(publish_all())

      # Muestreo para la gráfica en movimiento
      while not task.done() or sum(counters) < sent:
          await asyncio.sleep(0.08)
          history.append((time.monotonic() - t0, sent, sum(counters)))
          if live:
              render(history, n_subs, n_msgs)
          if time.monotonic() - t0 > 30:    # tope de seguridad
              break
      await task

      # Drenaje final (que los callbacks alcancen al publisher)
      for _ in range(40):
          if sum(counters) >= sent:
              break
          await asyncio.sleep(0.05)
      history.append((time.monotonic() - t0, sent, sum(counters)))
      if live:
          render(history, n_subs, n_msgs, done=True)

      for s in subs:
          await s.unsubscribe()
      await nc.drain()
      return history, counters

  def on_click(_):
      run_btn.disabled = True
      with log_out:
          clear_output()
          print(f"Lanzando: {n_msgs_w.value:,} mensajes → {n_subs_w.value} subscribers …")
      async def go():
          try:
              history, counters = await run_benchmark(n_msgs_w.value, n_subs_w.value, live=True)
              dur = history[-1][0]
              recv = sum(counters)
              with log_out:
                  print(f"OK en {dur:.2f}s")
                  print(f"  enviados : {n_msgs_w.value:,}")
                  print(f"  recibidos: {recv:,}  (fan-out ×{n_subs_w.value} = {recv/max(n_msgs_w.value,1):.2f} por mensaje)")
                  print(f"  throughput pub : {n_msgs_w.value/dur:,.0f} msgs/s")
                  print(f"  throughput recv: {recv/dur:,.0f} msgs/s (entregas totales)")
                  print(f"  por subscriber : {counters}")
          finally:
              run_btn.disabled = False
      asyncio.ensure_future(go())

  run_btn.on_click(on_click)
  display(widgets.HBox([n_msgs_w, n_subs_w]), run_btn, plot_out, log_out)
  print("Simulador listo. Pulsa el botón para lanzar el benchmark.")

source_raw:
  import ipywidgets as widgets
  from IPython.display import display, clear_output
  import matplotlib.pyplot as plt
  import asyncio, time
  import nats

  # --- widgets ---
  n_msgs_w = widgets.IntSlider(value=20000, min=1000, max=100000, step=1000,
                               description="Mensajes:", style={"description_width": "initial"},
                               layout=widgets.Layout(width="380px"))
  n_subs_w = widgets.IntSlider(value=3, min=1, max=8, step=1,
                               description="Subscribers:", style={"description_width": "initial"})
  run_btn  = widgets.Button(description="▶ Ejecutar benchmark", button_style="success",
                            layout=widgets.Layout(width="220px"))
  plot_out = widgets.Output()
  log_out  = widgets.Output()

  SUBJECT = "bench.load"
  PAYLOAD = b"x" * 128   # 128 bytes por mensaje

  def _throughput(ts, recv):
      thr = [0.0]
      for i in range(1, len(ts)):
          dt = ts[i] - ts[i-1]
          thr.append((recv[i] - recv[i-1]) / dt if dt > 0 else 0.0)
      return thr

  def render(history, n_subs, n_msgs, done=False):
      ts   = [h[0] for h in history]
      sent = [h[1] for h in history]
      recv = [h[2] for h in history]
      thr  = _throughput(ts, recv)
      with plot_out:
          clear_output(wait=True)
          fig, (a1, a2) = plt.subplots(1, 2, figsize=(11, 3.6))
          a1.plot(ts, sent, label="enviados (pub)", color="#2563eb", lw=2)
          a1.plot(ts, recv, label=f"recibidos (Σ {n_subs} subs)", color="#16a34a", lw=2)
          a1.set_xlabel("segundos"); a1.set_ylabel("mensajes acumulados")
          a1.set_title("Publisher vs subscribers"); a1.legend(loc="upper left")
          a2.plot(ts, thr, color="#db2777", lw=2)
          a2.set_xlabel("segundos"); a2.set_ylabel("msgs/s recibidos")
          a2.set_title("Throughput instantáneo")
          estado = "✓ DONE" if done else "● corriendo…"
          fig.suptitle(f"[{estado}]  {n_msgs:,} msgs → {n_subs} subs   "
                       f"enviados={sent[-1]:,}  recibidos={recv[-1]:,}", fontsize=11)
          plt.tight_layout(); plt.show()

  async def run_benchmark(n_msgs, n_subs, live=True):
      """1 publisher -> n_subs subscribers. Devuelve (history, counters)."""
      nc = await nats.connect(NATS_URL, name="benchmark")
      counters = [0] * n_subs

      def make_cb(i):
          async def cb(msg):
              counters[i] += 1
          return cb

      subs = [await nc.subscribe(SUBJECT, cb=make_cb(i)) for i in range(n_subs)]
      history = []          # (t, enviados, recibidos_total)
      sent = 0
      t0 = time.monotonic()

      async def publish_all():
          nonlocal sent
          for k in range(n_msgs):
              await nc.publish(SUBJECT, PAYLOAD)
              sent += 1
              if k % 1000 == 0:
                  await nc.flush()
                  await asyncio.sleep(0)   # ceder al event loop (deja correr callbacks)
          await nc.flush()

      task = asyncio.create_task(publish_all())

      # Muestreo para la gráfica en movimiento
      while not task.done() or sum(counters) < sent:
          await asyncio.sleep(0.08)
          history.append((time.monotonic() - t0, sent, sum(counters)))
          if live:
              render(history, n_subs, n_msgs)
          if time.monotonic() - t0 > 30:    # tope de seguridad
              break
      await task

      # Drenaje final (que los callbacks alcancen al publisher)
      for _ in range(40):
          if sum(counters) >= sent:
              break
          await asyncio.sleep(0.05)
      history.append((time.monotonic() - t0, sent, sum(counters)))
      if live:
          render(history, n_subs, n_msgs, done=True)

      for s in subs:
          await s.unsubscribe()
      await nc.drain()
      return history, counters

  def on_click(_):
      run_btn.disabled = True
      with log_out:
          clear_output()
          print(f"Lanzando: {n_msgs_w.value:,} mensajes → {n_subs_w.value} subscribers …")
      async def go():
          try:
              history, counters = await run_benchmark(n_msgs_w.value, n_subs_w.value, live=True)
              dur = history[-1][0]
              recv = sum(counters)
              with log_out:
                  print(f"OK en {dur:.2f}s")
                  print(f"  enviados : {n_msgs_w.value:,}")
                  print(f"  recibidos: {recv:,}  (fan-out ×{n_subs_w.value} = {recv/max(n_msgs_w.value,1):.2f} por mensaje)")
                  print(f"  throughput pub : {n_msgs_w.value/dur:,.0f} msgs/s")
                  print(f"  throughput recv: {recv/dur:,.0f} msgs/s (entregas totales)")
                  print(f"  por subscriber : {counters}")
          finally:
              run_btn.disabled = False
      asyncio.ensure_future(go())

  run_btn.on_click(on_click)
  display(widgets.HBox([n_msgs_w, n_subs_w]), run_btn, plot_out, log_out)
  print("Simulador listo. Pulsa el botón para lanzar el benchmark.")


source:
  hist, counters = await run_benchmark(15000, 3, live=False)

  ts   = [h[0] for h in hist]
  sent = [h[1] for h in hist]
  recv = [h[2] for h in hist]
  dur  = ts[-1]

  fig, ax = plt.subplots(figsize=(9, 3.6))
  ax.plot(ts, sent, label="enviados (pub)", color="#2563eb", lw=2)
  ax.plot(ts, recv, label="recibidos (Σ 3 subs)", color="#16a34a", lw=2)
  ax.set_xlabel("segundos"); ax.set_ylabel("mensajes acumulados")
  ax.set_title(f"Benchmark headless: 15.000 msgs → 3 subs en {dur:.2f}s")
  ax.legend(loc="upper left")
  plt.tight_layout(); plt.show()

  print(f"enviados : 15,000")
  print(f"recibidos: {sum(counters):,}  por sub -> {counters}")
  print(f"throughput pub : {15000/dur:,.0f} msgs/s")
  print(f"throughput recv: {sum(counters)/dur:,.0f} msgs/s (entregas totales, fan-out x3)")

source_raw:
  hist, counters = await run_benchmark(15000, 3, live=False)

  ts   = [h[0] for h in hist]
  sent = [h[1] for h in hist]
  recv = [h[2] for h in hist]
  dur  = ts[-1]

  fig, ax = plt.subplots(figsize=(9, 3.6))
  ax.plot(ts, sent, label="enviados (pub)", color="#2563eb", lw=2)
  ax.plot(ts, recv, label="recibidos (Σ 3 subs)", color="#16a34a", lw=2)
  ax.set_xlabel("segundos"); ax.set_ylabel("mensajes acumulados")
  ax.set_title(f"Benchmark headless: 15.000 msgs → 3 subs en {dur:.2f}s")
  ax.legend(loc="upper left")
  plt.tight_layout(); plt.show()

  print(f"enviados : 15,000")
  print(f"recibidos: {sum(counters):,}  por sub -> {counters}")
  print(f"throughput pub : {15000/dur:,.0f} msgs/s")
  print(f"throughput recv: {sum(counters)/dur:,.0f} msgs/s (entregas totales, fan-out x3)")


