o ^۷i0@@sddlZddlZddlZddlZddlmZmZddlmZmZddl m Z m Z m Z m Z mZmZmZmZmZmZddlZddlmZddlmZddlmZddlmZdd lmZmZm Z m!Z!dd l"m#Z#m$Z$e%e&Z'eGd d d Z(e)Z*d e(fddZ+ddZ,dddZ-dddZ.Gddde/Z0GdddZ1zddl2m3Z3e3j4e1_5e1j4e3_4Wne6ye'7dYnwee8e9e:e#e fZ;ee;ZGdddeZ?dS)N) dataclassfield)datetimetimezone) Any AsyncIterable AwaitableCallable CoroutineIteratorMappingOptionalSetUnion)BackgroundTask)iterate_in_threadpool)MutableHeaders)Response)ReceiveScopeSendMessage)ServerSentEvent ensure_bytesc@s6eZdZUdZeedZeej e d<dZ e e d<dS)_ShutdownStatezPer-thread state for shutdown coordination. Issue #152 fix: Uses threading.local() instead of ContextVar to ensure one watcher per thread rather than one per async context. )default_factoryeventsFwatcher_startedN) __name__ __module__ __qualname____doc__rsetrranyioEvent__annotations__rboolr'r'G/home/ubuntu/vllm_env/lib/python3.10/site-packages/sse_starlette/sse.pyr!s rreturncCs$ttdd}|durt}|t_|S)z4Get or create shutdown state for the current thread.shutdown_stateN)getattr _thread_staterr*)stater'r'r(_get_shutdown_state1s r.cCsNzttj}t|dr|j}t|dr|WSWdSWdSty&YdSw)am Try to get uvicorn Server instance via signal handler introspection. When uvicorn registers signal handlers, they're bound methods on the Server instance. We can retrieve the Server from the handler's __self__ attribute. Returns None if: - Not running under uvicorn - Signal handler isn't a bound method - Any introspection fails __self__ should_exitN)signal getsignalSIGTERMhasattrr/ Exception)handlerserverr'r'r(_get_uvicorn_server:s    r8cstt}t}z. tjr ntjr|dur|jrdt_n tdIdHq t|jD]}| q)Wd|_ dSd|_ w)ag Poll for shutdown and broadcast to all events in this context. One watcher runs per thread (event loop). Checks two shutdown sources: 1. AppStatus.should_exit - set when our monkey-patch works 2. uvicorn Server.should_exit - via signal handler introspection (Issue #132 fix) When either becomes True, signals all registered events. TNg?F) r.r8 AppStatusr0enable_automatic_graceful_drainr#sleeplistrr"r)r-uvicorn_servereventr'r'r(_shutdown_watcherQs(  r?cCsLt}|js$d|_z t}|tWdSty#d|_YdSwdS)zDEnsure the shutdown watcher is running for this thread (event loop).TFN)r.rasyncioget_running_loop create_taskr? RuntimeError)r-loopr'r'r($_ensure_watcher_started_on_this_loopus  rEc@s eZdZdS)SendTimeoutErrorN)rrr r'r'r'r(rFsrFc@sNeZdZUdZdZdZdZeee d<e ddZ e dd Z e d d Z dS) r9z\Helper to capture a shutdown signal from Uvicorn so we can gracefully terminate SSE streams.FTNoriginal_handlercC dt_dS)aJ Prevent automatic SSE stream termination on server shutdown. WARNING: When disabled, you MUST set AppStatus.should_exit = True at some point during shutdown, or streams will never close and the server will hang indefinitely (or until uvicorn's graceful shutdown timeout expires). FNr9r:r'r'r'r( disable_automatic_graceful_drains z*AppStatus.disable_automatic_graceful_draincCrH)a Re-enable automatic SSE stream termination on server shutdown. This restores the default behavior where SIGTERM triggers immediate stream draining. Call this to undo a previous call to disable_automatic_graceful_drain(). TNrIr'r'r'r($enable_automatic_graceful_drain_modes z.AppStatus.enable_automatic_graceful_drain_modecOs.tjrdt_tjdurtj|i|dSdS)NT)r9r:r0rG)argskwargsr'r'r( handle_exits  zAppStatus.handle_exit)rrr r!r0r:rGr r r% staticmethodrJrKrNr'r'r'r(r9s    r9)ServerzHUvicorn not installed. Graceful shutdown on server termination disabled.c@szeZdZdZdZdZ            d0ded ed ee e e fd e d ee d eedee dee ge fdee gedfdeedee egedfdeejdeddfddZedeeeffddZejdeeefddfddZd1deddfdd Zd!eddfd"d#Zd$eddfd%d&Zed2d'd(Z d2d)d*Z!d!eddfd+d,Z"d-e#d$ed!eddfd.d/Z$dS)3EventSourceResponseagStreaming response implementing the SSE (Server-Sent Events) specification. Args: content: Async iterable or sync iterator yielding SSE event data. status_code: HTTP status code. Default: 200. headers: Additional HTTP headers. media_type: Response media type. Default: "text/event-stream". background: Background task to run after response completes. ping: Ping interval in seconds (0 to disable). Default: 15. sep: Line separator for SSE messages ("\r\n", "\r", or "\n"). ping_message_factory: Callable returning custom ping ServerSentEvent. data_sender_callable: Async callable for push-based data sending. send_timeout: Timeout in seconds for individual send operations. client_close_handler_callable: Async callback on client disconnect. shutdown_event: Optional ``anyio.Event`` set by the library when server shutdown is detected. Generators can watch this event to send farewell messages and exit cooperatively instead of receiving CancelledError. shutdown_grace_period: Seconds to wait after setting ``shutdown_event`` before force-cancelling the generator. Must be >= 0. Should be less than your ASGI server's graceful shutdown timeout. Default: 0 (immediate cancel, identical to pre-v3.3.0 behavior).  Ntext/event-streamrcontent status_codeheaders media_type backgroundpingsepping_message_factorydata_sender_callable)NNN send_timeoutclient_close_handler_callableshutdown_eventshutdown_grace_periodr)cCs|dvr td||p|j|_t|tr||_nt||_||_|dur)|jn||_||_ | |_ | |_ t }|durA| ||ddd|d<d|d<|||dur[|jn||_||_| |_| d krltd | |_| |_d |_t|_dS) N)NrS  z'sep must be one of: \r\n, \r, \n, got: z Cache-Controlzno-storez keep-alive ConnectionnozX-Accel-Bufferingrz"shutdown_grace_period must be >= 0T) ValueErrorDEFAULT_SEPARATORr\ isinstancer body_iteratorrrWrYrZr^r_rupdate setdefault init_headersDEFAULT_PING_INTERVAL ping_intervalr]r`_shutdown_event_shutdown_grace_periodactiver#Lock _send_lock)selfrVrWrXrYrZr[r\r]r^r_r`rarb_headersr'r'r(__init__s6      zEventSourceResponse.__init__cCs|jSN)_ping_intervalrur'r'r(ro sz!EventSourceResponse.ping_intervalvaluecCs0t|ttfs td|dkrtd||_dS)Nzping interval must be intrz$ping interval must be greater than 0)riintfloat TypeErrorrgry)rur{r'r'r(ro$s  FforcecCstd)Nz-Compression is not supported for SSE streams.)NotImplementedError)rurr'r'r(enable_compression,sz&EventSourceResponse.enable_compressionsendc s|d|j|jdIdH|j2zL3dHW}t||j}td|t|j }|d|ddIdHWdn1s?wY|r]|j r]t |jdd}|durZ|IdHt q6|j 4IdHd |_|dd d dIdHWdIdHdS1IdHswYdS) zHSend out SSE data to the client as it becomes available in the iterator.zhttp.response.start)typestatusrXNz chunk: %shttp.response.bodyTrbody more_bodyacloseF)rW raw_headersrjrr\loggerdebugr# move_on_afterr_ cancel_calledr+rFrtrr)rurdatachunk cancel_scoperr'r'r(_stream_response/s4       .z$EventSourceResponse._stream_responsereceivecsX|jr*|IdH}|ddkr%d|_td|jr#||IdHdS|jsdSdS)z/Watch for a disconnect message from the client.Nrzhttp.disconnectFz+Got event: http.disconnect. Stop streaming.)rrrrr`)rurmessager'r'r(_listen_for_disconnectKs   z*EventSourceResponse._listen_for_disconnectc sttjrdStt}t}|j|ztjr#W|j|dS| IdHW|j|dS|j|w)z0Wait for shutdown signal via the shared watcher.N) r9r0rEr.r#r$radddiscardwait)r-r>r'r'r(_listen_for_exit_signalVs z+EventSourceResponse._listen_for_exit_signalcs|IdH|jr|j|jdkrDt|j|jr2tdIdH|jsWddSWddS1s=wYdSdS)a~Wait for shutdown signal, then optionally give generator a grace period. Issue #167: When a shutdown_event is provided, the library sets it before returning, giving the generator a chance to send farewell events and exit cooperatively. The shutdown_grace_period controls how long to wait before force-cancelling via task group cancellation. Nrg?)rrpr"rqr#rrrr;rzr'r'r("_listen_for_exit_signal_with_gracejs  "z6EventSourceResponse._listen_for_exit_signal_with_gracec s|jr^t|jIdH|jr|n tdttj |j d}t ||j }t d||j4IdH|jrD|d|ddIdHWdIdHn 1IdHsTwY|jsdSdS)zPeriodically send ping messages to keep the connection alive on proxies. - frequenccy ca every 15 seconds. - Alternatively one can send periodically a comment line (one starting with a ':' character) Nzping - )commentr\zping: %srTr)rrr#r;ryr]rrnowrutcr\rrrrt)rursse_ping ping_bytesr'r'r(_ping~s,    (zEventSourceResponse._pingscopec st4IdHKdtgtdfffdd }|fdd|fdd|jjr?j|fddWdIdHn 1IdHsZwYjdurmIdHdSdS) aEntrypoint for Starlette's ASGI contract. We spin up tasks: - _stream_response to push events - _ping to keep the connection alive - _listen_for_exit_signal to respond to server shutdown - _listen_for_disconnect to respond to client disconnect Ncorocs|IdHjdSrx)rcancel)r) task_groupr'r(cancel_on_finishs z6EventSourceResponse.__call__..cancel_on_finishc Srx)rr'rurr'r( z.EventSourceResponse.__call__..crrx)rr'rr'r(rrcs Srx)rr')rrur'r(rr)r#create_task_groupr r start_soonrr^rZ)rurrrrr')rrurrr(__call__s" ( zEventSourceResponse.__call__) rTNrUNNNNNNNNr)Fr)N)%rrr r!rnrh ContentStreamr|r r strrr rr r}rrr#r$rwpropertyrrosetterr&rrrrrrOrrrrrr'r'r'r(rQsz    I  rQr)@r@loggingr1 threading dataclassesrrrrtypingrrrr r r r r rrr#starlette.backgroundrstarlette.concurrencyrstarlette.datastructuresrstarlette.responsesrstarlette.typesrrrrsse_starlette.eventrr getLoggerrrrlocalr,r.r8r?rE TimeoutErrorrFr9 uvicorn.mainrPrNrG ImportErrorrrbytesdictContentSyncContentStreamAsyncContentStreamrrQr'r'r'r(sL0        $ &