#!/usr/bin/env python3 """ Comprehensive test application demonstrating the fix for issue #132. This application proves that the enhanced AppStatus implementation correctly handles shutdown signals with uvicorn 0.34+, replacing the broken monkey-patching approach with direct signal handler registration. Run with: python issue132_fix_demo.py Test with: Open http://localhost:8080 and press Ctrl+C to see graceful shutdown """ import asyncio import logging import signal import threading import time from datetime import datetime, timezone from typing import AsyncGenerator, Dict, List, Any import uvicorn from fastapi import FastAPI, BackgroundTasks from fastapi.responses import HTMLResponse from pydantic import BaseModel from starlette.middleware.cors import CORSMiddleware from starlette.requests import Request from sse_starlette import EventSourceResponse from sse_starlette.appstatus import AppStatus # Configure comprehensive logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)8s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) logger = logging.getLogger("issue132_demo") # Global state for demonstration class DemoState: """Track demo state to show proper cleanup.""" active_streams: int = 0 total_events_sent: int = 0 startup_time: float = time.time() shutdown_initiated: bool = False cleanup_completed: bool = False demo_state = DemoState() class StreamStats(BaseModel): """Response model for stream statistics.""" active_streams: int total_events_sent: int uptime_seconds: float shutdown_initiated: bool def create_shutdown_callbacks() -> None: """Register shutdown callbacks to demonstrate the fix.""" def log_shutdown_initiation(): logger.info("๐Ÿ›‘ Shutdown callback #1: Shutdown initiated") demo_state.shutdown_initiated = True def cleanup_resources(): logger.info("๐Ÿงน Shutdown callback #2: Cleaning up resources") # Simulate resource cleanup time.sleep(0.1) demo_state.cleanup_completed = True def final_shutdown_log(): logger.info("โœ… Shutdown callback #3: Final cleanup complete") logger.info(f"๐Ÿ“Š Final stats: {demo_state.active_streams} streams, " f"{demo_state.total_events_sent} events sent") # Register callbacks in order AppStatus.add_shutdown_callback(log_shutdown_initiation) AppStatus.add_shutdown_callback(cleanup_resources) AppStatus.add_shutdown_callback(final_shutdown_log) logger.info("๐Ÿ“ Registered 3 shutdown callbacks") async def demonstrate_signal_handling() -> AsyncGenerator[Dict, None]: """ Stream that demonstrates proper signal handling. This will terminate gracefully when SIGINT/SIGTERM is received. """ demo_state.active_streams += 1 event_count = 0 try: logger.info(f"๐Ÿš€ Starting demo stream (active streams: {demo_state.active_streams})") while not AppStatus.should_exit: event_count += 1 demo_state.total_events_sent += 1 # Generate different types of events to show variety if event_count % 10 == 0: event_type = "milestone" message = f"Milestone reached: {event_count} events sent" elif event_count % 5 == 0: event_type = "status" message = f"Status update: {demo_state.active_streams} active streams" else: event_type = "data" message = f"Event #{event_count} at {datetime.now().strftime('%H:%M:%S')}" yield { "event": event_type, "data": { "message": message, "event_id": event_count, "timestamp": time.time(), "uptime": time.time() - demo_state.startup_time, "should_exit": AppStatus.should_exit }, "id": str(event_count) } # Check for shutdown more frequently than sleep interval for _ in range(10): # Check 10 times per second if AppStatus.should_exit: logger.info(f"๐Ÿ”„ Stream {id(asyncio.current_task())} received shutdown signal") break await asyncio.sleep(0.1) except asyncio.CancelledError: logger.info(f"โŒ Stream cancelled after {event_count} events") raise except Exception as e: logger.error(f"๐Ÿ’ฅ Stream error after {event_count} events: {e}") raise finally: demo_state.active_streams -= 1 logger.info(f"๐Ÿ Stream ended. Sent {event_count} events. " f"Active streams: {demo_state.active_streams}") async def health_monitoring_stream() -> AsyncGenerator[Dict, None]: """ Health monitoring stream to show multiple concurrent SSE endpoints. """ demo_state.active_streams += 1 check_count = 0 try: logger.info("๐Ÿฅ Starting health monitoring stream") while not AppStatus.should_exit: check_count += 1 yield { "event": "health_check", "data": { "status": "healthy" if not AppStatus.should_exit else "shutting_down", "check_number": check_count, "active_streams": demo_state.active_streams, "total_events": demo_state.total_events_sent, "uptime": time.time() - demo_state.startup_time, "memory_info": "simulated_memory_stats", "app_status": { "should_exit": AppStatus.should_exit, "initialized": AppStatus._initialized, "callbacks_registered": len(AppStatus._shutdown_callbacks) } }, "id": f"health_{check_count}" } # Health checks every 3 seconds, but check shutdown more frequently for _ in range(30): # 0.1s * 30 = 3s total, but responsive to shutdown if AppStatus.should_exit: break await asyncio.sleep(0.1) except asyncio.CancelledError: logger.info(f"๐Ÿฅ Health monitoring cancelled after {check_count} checks") raise finally: demo_state.active_streams -= 1 logger.info(f"๐Ÿ Health monitoring ended after {check_count} checks") # FastAPI Application Setup app = FastAPI( title="Issue #132 Fix Demonstration", description="Demonstrates the enhanced AppStatus signal handling fix", version="1.0.0" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/", response_class=HTMLResponse) async def homepage() -> str: """Serve the test page.""" return """ Issue #132 Fix Demonstration

๐Ÿ”ง Issue #132 Fix Demonstration

Enhanced AppStatus with Direct Signal Handler Registration

uvicorn 0.34+ Compatible - No more monkey-patching!

๐Ÿงช How to Test the Fix

  1. Start Streams: Click the buttons below to start SSE streams
  2. Monitor Activity: Watch the real-time events and statistics
  3. Test Shutdown: Press Ctrl+C in the terminal
  4. Observe Cleanup: Notice graceful shutdown in terminal logs
  5. Verify Fix: No hanging processes or zombie connections

๐ŸŽ›๏ธ Stream Controls

0
Active Streams
0
Total Events
0s
Uptime
Normal
Status

๐Ÿ“ก Live Event Streams

Main Stream: Disconnected
Health Stream: Disconnected
""" @app.get("/events") async def main_event_stream(request: Request) -> EventSourceResponse: """Main SSE endpoint demonstrating the signal handling fix.""" logger.info(f"๐Ÿ”Œ New main stream connection from {request.client}") return EventSourceResponse( demonstrate_signal_handling(), headers={ "X-Demo-Purpose": "issue-132-fix", "X-Connection-ID": str(id(request)), }, ping=15 # Ping every 15 seconds ) @app.get("/health") async def health_stream(request: Request) -> EventSourceResponse: """Health monitoring SSE endpoint.""" logger.info(f"๐Ÿฅ New health monitoring connection from {request.client}") return EventSourceResponse( health_monitoring_stream(), headers={ "X-Stream-Type": "health-monitoring", "X-Connection-ID": str(id(request)), }, ping=30 # Less frequent pings for health monitoring ) @app.get("/stats") async def get_statistics() -> StreamStats: """Get current statistics to show in the UI.""" return StreamStats( active_streams=demo_state.active_streams, total_events_sent=demo_state.total_events_sent, uptime_seconds=time.time() - demo_state.startup_time, shutdown_initiated=demo_state.shutdown_initiated ) @app.get("/test-signal") async def test_signal_handling() -> Dict[str, Any]: """ Test endpoint to programmatically trigger shutdown handling. This allows testing without sending actual signals. """ logger.info("๐Ÿงช Testing signal handling programmatically") # Simulate signal handling AppStatus.handle_exit(signal.SIGTERM, None) return { "message": "Signal handling test triggered", "should_exit": AppStatus.should_exit, "shutdown_callbacks_count": len(AppStatus._shutdown_callbacks), "app_status_initialized": AppStatus._initialized } @app.on_event("startup") async def startup_event(): """Application startup handler.""" logger.info("๐Ÿš€ Application startup") logger.info(f"๐Ÿ“‹ AppStatus initialized: {AppStatus._initialized}") logger.info(f"๐Ÿ”ง Signal handlers registered: {len(AppStatus._original_handlers)}") # Register our demonstration callbacks create_shutdown_callbacks() # Log the fix details logger.info("โœ… Issue #132 Fix Active:") logger.info(" - Direct signal handler registration (no monkey-patching)") logger.info(" - Thread-safe shutdown handling") logger.info(" - Graceful SSE stream termination") logger.info(" - Compatible with uvicorn 0.34+") @app.on_event("shutdown") async def shutdown_event(): """Application shutdown handler.""" logger.info("๐Ÿ›‘ Application shutdown initiated") if demo_state.cleanup_completed: logger.info("โœ… Cleanup completed successfully") else: logger.warning("โš ๏ธ Cleanup may not have completed") def main(): """Main entry point with comprehensive logging.""" print("=" * 80) print("๐Ÿ”ง ISSUE #132 FIX DEMONSTRATION") print("=" * 80) print() print("This application demonstrates the enhanced AppStatus implementation") print("that fixes the signal handling issue with uvicorn 0.34+") print() print("Key improvements:") print(" โœ… Direct signal handler registration (no monkey-patching)") print(" โœ… Thread-safe shutdown handling") print(" โœ… Graceful SSE stream termination") print(" โœ… Backward compatibility maintained") print() print("๐Ÿงช Test Instructions:") print(" 1. Open http://localhost:8080 in your browser") print(" 2. Start some SSE streams using the web interface") print(" 3. Press Ctrl+C in this terminal to test graceful shutdown") print(" 4. Observe clean termination and proper cleanup in logs") print() print("๐Ÿ“Š Monitor the logs below for detailed shutdown behavior...") print("=" * 80) print() # Configure uvicorn for optimal demonstration config = uvicorn.Config( app=app, host="0.0.0.0", port=8080, log_level="info", access_log=True, reload=False, # Disable reload to see clean shutdown behavior workers=1, # Single worker for clear demonstration ) server = uvicorn.Server(config) try: server.run() except KeyboardInterrupt: logger.info("๐Ÿ›‘ KeyboardInterrupt received - testing graceful shutdown") except Exception as e: logger.error(f"๐Ÿ’ฅ Unexpected error: {e}") finally: # Verify fix worked if demo_state.shutdown_initiated and demo_state.cleanup_completed: print("\n" + "=" * 80) print("โœ… SUCCESS: Issue #132 fix verified!") print(" - Shutdown callbacks executed") print(" - Resources cleaned up properly") print(" - No hanging processes") print("=" * 80) else: print("\n" + "=" * 80) print("โŒ Issue #132 fix may have problems") print(f" - Shutdown initiated: {demo_state.shutdown_initiated}") print(f" - Cleanup completed: {demo_state.cleanup_completed}") print("=" * 80) if __name__ == "__main__": main()