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

@@ -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 ])

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)

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'

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