Revert "attempting proxying"
This reverts commit 7d23bf60cf.
changed to 1 worker
This commit is contained in:
@@ -3,7 +3,6 @@ import queue
|
||||
import uuid
|
||||
from typing import Dict, Any
|
||||
import logging
|
||||
from multiprocessing import Manager
|
||||
|
||||
from flask import Response
|
||||
|
||||
@@ -11,32 +10,38 @@ from events.types.event import Event
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
manager = Manager()
|
||||
# Maps user_id → dict of {connection_id: queue}
|
||||
user_queues = manager.dict() # Shared across all Gunicorn workers
|
||||
user_queues: Dict[str, Dict[str, queue.Queue]] = {}
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
|
||||
def get_queue(user_id: str, connection_id: str) -> queue.Queue:
|
||||
"""Get or create a queue for a specific user connection."""
|
||||
if user_id not in user_queues:
|
||||
user_queues[user_id] = manager.dict()
|
||||
user_queues[user_id] = {}
|
||||
|
||||
if connection_id not in user_queues[user_id]:
|
||||
user_queues[user_id][connection_id] = manager.Queue()
|
||||
user_queues[user_id][connection_id] = queue.Queue()
|
||||
|
||||
return user_queues[user_id][connection_id]
|
||||
|
||||
|
||||
def send_to_user(user_id: str, data: Dict[str, Any]):
|
||||
"""Send data to all connections for a specific user."""
|
||||
logger.info(f"Sending data to {user_id}")
|
||||
logger.info(f"Sending data to {user_id} user quesues are {user_queues.keys()}")
|
||||
if user_id in user_queues:
|
||||
logger.info(f"Queued {user_id}")
|
||||
# Format as SSE message once
|
||||
message = f"data: {json.dumps(data)}\n\n".encode('utf-8')
|
||||
|
||||
# Send to all connections for this user
|
||||
for connection_id, q in user_queues[user_id].items():
|
||||
try:
|
||||
logger.info(f"Sending message to {connection_id}")
|
||||
q.put(message)
|
||||
q.put(message, block=False)
|
||||
except queue.Full:
|
||||
# Skip if queue is full (connection might be dead)
|
||||
pass
|
||||
|
||||
|
||||
@@ -58,7 +63,6 @@ def sse_response_for_user(user_id: str):
|
||||
# Get message from queue (blocks until available)
|
||||
logger.info(f"blocking on get for {user_id} user")
|
||||
message = user_queue.get()
|
||||
logger.info(f"-------received message {message}")
|
||||
yield message
|
||||
except GeneratorExit:
|
||||
# Clean up when client disconnects
|
||||
@@ -78,9 +82,3 @@ def sse_response_for_user(user_id: str):
|
||||
'Connection': 'keep-alive'
|
||||
}
|
||||
)
|
||||
|
||||
_shared_manager = None
|
||||
|
||||
def set_shared_manager(manager):
|
||||
global _shared_manager
|
||||
_shared_manager = manager
|
||||
|
||||
Reference in New Issue
Block a user