40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
|
|
import asyncio
|
|
import os
|
|
import nats
|
|
|
|
async def main():
|
|
# Use environment variables for connection logic to match the worker pod's context
|
|
nats_url = os.getenv("NATS_URL", "nats://nats-server:4222")
|
|
nats_user = os.getenv("NATS_USER", "admin")
|
|
nats_password = os.getenv("NATS_PASSWORD", "package@321#")
|
|
|
|
print(f"Connecting to NATS at {nats_url}...")
|
|
|
|
try:
|
|
nc = await nats.connect(nats_url, user=nats_user, password=nats_password)
|
|
js = nc.jetstream()
|
|
print("Connected!")
|
|
|
|
# List of streams to clean
|
|
streams = ["CUSTOMERS", "DELIVERIES", "ORDERS", "RIDER", "PRODUCTS"]
|
|
|
|
for stream in streams:
|
|
print(f"Checking stream: {stream}...")
|
|
try:
|
|
consumers = await js.consumers_info(stream)
|
|
for c in consumers:
|
|
# We want to delete the stuck consumers to let them be recreated properly
|
|
print(f" - Deleting consumer: {c.name}")
|
|
await js.delete_consumer(stream, c.name)
|
|
except Exception as e:
|
|
print(f" Note: Stream {stream} check skipped/failed: {e}")
|
|
|
|
await nc.close()
|
|
|
|
except Exception as e:
|
|
print(f"Fatal connection error: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|