From 48869bedf080271d2a41294926b91b6201f7c792 Mon Sep 17 00:00:00 2001 From: Henry Ruhs Date: Tue, 19 May 2026 17:05:53 +0200 Subject: [PATCH] Follow WHIP specs (#1123) * follow more specs of whip * pass the location header value via API router * fix CI * remove more queries --- facefusion/apis/core.py | 3 ++- facefusion/apis/endpoints/stream.py | 36 +++++++++++++++++------------ facefusion/apis/stream_helper.py | 8 +++++++ tests/test_api_stream.py | 35 ++++++++++++++++++++++++++-- 4 files changed, 64 insertions(+), 18 deletions(-) diff --git a/facefusion/apis/core.py b/facefusion/apis/core.py index effd7341..4f59ba89 100644 --- a/facefusion/apis/core.py +++ b/facefusion/apis/core.py @@ -9,7 +9,7 @@ from facefusion.apis.endpoints.metrics import get_metrics, websocket_metrics from facefusion.apis.endpoints.ping import websocket_ping from facefusion.apis.endpoints.session import create_session, destroy_session, get_session, refresh_session from facefusion.apis.endpoints.state import get_state, set_state -from facefusion.apis.endpoints.stream import post_stream, websocket_stream +from facefusion.apis.endpoints.stream import delete_stream, post_stream, websocket_stream from facefusion.apis.middlewares.session import create_session_guard @@ -30,6 +30,7 @@ def create_api() -> Starlette: Route('/capabilities', get_capabilities, methods = [ 'GET' ]), Route('/metrics', get_metrics, methods = [ 'GET' ], middleware = [ session_guard ]), Route('/stream', post_stream, methods = [ 'POST' ], middleware = [ session_guard ]), + Route('/stream', delete_stream, methods = [ 'DELETE' ], name = 'delete_stream', middleware = [ session_guard ]), WebSocketRoute('/metrics', websocket_metrics, middleware = [ session_guard ]), WebSocketRoute('/ping', websocket_ping, middleware = [ session_guard ]), WebSocketRoute('/stream', websocket_stream, middleware = [ session_guard ]) diff --git a/facefusion/apis/endpoints/stream.py b/facefusion/apis/endpoints/stream.py index 207a7be6..233374f6 100644 --- a/facefusion/apis/endpoints/stream.py +++ b/facefusion/apis/endpoints/stream.py @@ -1,36 +1,42 @@ from starlette.requests import Request from starlette.responses import Response -from starlette.status import HTTP_201_CREATED, HTTP_404_NOT_FOUND +from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_404_NOT_FOUND from starlette.websockets import WebSocket from facefusion import session_context, session_manager from facefusion.apis.session_helper import extract_access_token -from facefusion.apis.stream_helper import process_image, process_video +from facefusion.apis.stream_helper import destroy_stream, process_image, process_video async def websocket_stream(websocket : WebSocket) -> None: - stream_type = websocket.query_params.get('type') - - if stream_type == 'image': - return await process_image(websocket) - - return await websocket.close(1008) + return await process_image(websocket) async def post_stream(request : Request) -> Response: - stream_type = request.query_params.get('type') + headers =\ + { + 'Location': request.url_for('delete_stream').path + } content_type = request.headers.get('content-type') access_token = extract_access_token(request.scope) session_id = session_manager.find_session_id(access_token) - session_context.set_session_id(session_id) - if content_type == 'application/sdp' and session_id: + if session_id and content_type == 'application/sdp': sdp_offer = await request.body() + sdp_answer = process_video(session_id, sdp_offer.decode()) - if stream_type == 'video': - sdp_answer = process_video(session_id, sdp_offer.decode()) - - return Response(sdp_answer, status_code = HTTP_201_CREATED, media_type = 'application/sdp') + if sdp_answer: + return Response(sdp_answer, status_code = HTTP_201_CREATED, media_type = 'application/sdp', headers = headers) + + return Response(status_code = HTTP_404_NOT_FOUND) + + +async def delete_stream(request : Request) -> Response: + access_token = extract_access_token(request.scope) + session_id = session_manager.find_session_id(access_token) + + if session_id and destroy_stream(session_id): + return Response(status_code = HTTP_200_OK) return Response(status_code = HTTP_404_NOT_FOUND) diff --git a/facefusion/apis/stream_helper.py b/facefusion/apis/stream_helper.py index 6495defb..620a2c0b 100644 --- a/facefusion/apis/stream_helper.py +++ b/facefusion/apis/stream_helper.py @@ -19,6 +19,7 @@ from facefusion.types import AomDecoder, AomEncoder, AudioCodec, AudioFrame, Opu #TODO: needs review async def process_image(websocket : WebSocket) -> None: + #TODO: all the websocket handling belongs to the endpoint, these are connection concerns subprotocol = get_sec_websocket_protocol(websocket.scope) access_token = extract_access_token(websocket.scope) session_id = session_manager.find_session_id(access_token) @@ -55,6 +56,13 @@ async def receive_vision_frames(websocket : WebSocket) -> AsyncIterator[VisionFr websocket_event = await websocket.receive() +#TODO: just exist as endpoint stream.py is not allowed to access rtc store directly +def destroy_stream(session_id : SessionId) -> bool: + rtc_store.delete_peers(session_id) + + return not rtc_store.get_peers(session_id) + + #TODO: needs review def process_video(session_id : SessionId, sdp_offer : SdpOffer) -> Optional[SdpAnswer]: video_codec : VideoCodec = 'vp8' diff --git a/tests/test_api_stream.py b/tests/test_api_stream.py index 804cc783..6899e5c7 100644 --- a/tests/test_api_stream.py +++ b/tests/test_api_stream.py @@ -75,7 +75,7 @@ def test_stream_image(test_client : TestClient) -> None: assert select_response.status_code == 200 - with test_client.websocket_connect('/stream?type=image&action=process', subprotocols = + with test_client.websocket_connect('/stream', subprotocols = [ 'access_token.' + access_token ]) as websocket: @@ -126,7 +126,7 @@ def test_stream_video(test_client : TestClient, video_codec : VideoCodec) -> Non datachannel_module.create_static_library().rtcDeletePeerConnection(peer_connection) with patch('facefusion.rtc.send_video'): - stream_response = test_client.post('/stream?type=video&action=process', content = sdp_offer, headers = + stream_response = test_client.post('/stream', content = sdp_offer, headers = { 'Authorization': 'Bearer ' + access_token, 'Content-Type': 'application/sdp' @@ -134,3 +134,34 @@ def test_stream_video(test_client : TestClient, video_codec : VideoCodec) -> Non assert stream_response.status_code == 201 assert 'm=video' in stream_response.text + + +def test_delete_stream_video(test_client : TestClient) -> None: + create_session_response = test_client.post('/session', json = + { + 'client_version': metadata.get('version') + }) + access_token = create_session_response.json().get('access_token') + session_id = session_manager.find_session_id(access_token) + + peer_connection = rtc.create_peer_connection() + rtc.add_video_track(peer_connection, 'sendrecv', 'vp8', 96) + rtc.add_audio_track(peer_connection, 'sendrecv', 'opus', 111) + sdp_offer = rtc.create_sdp_offer(peer_connection) + datachannel_module.create_static_library().rtcDeletePeerConnection(peer_connection) + + test_client.post('/stream', content = sdp_offer, headers = + { + 'Authorization': 'Bearer ' + access_token, + 'Content-Type': 'application/sdp' + }) + + assert rtc_store.get_peers(session_id) + + delete_response = test_client.delete('/stream', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + + assert delete_response.status_code == 200 + assert rtc_store.get_peers(session_id) is None