decode_server_notifications

Author: roznawsk

examples/room_manager/routes.py

@@ -3,7 +3,7 @@
 from flask import Flask, abort, jsonify, request
 from room_service import RoomService
 
-from fishjam import receive_binary
+from fishjam import decode_server_notifications
 from fishjam.room import RoomType
 
 
@@ -36,12 +36,7 @@ def get_room_query():
 
     @app.post("/api/rooms/webhook")
     def webhook():
-        message = receive_binary(request.data)
-
-        if message:
-            notifications = message if isinstance(message, list) else [message]
-
-            for notification in notifications:
-                room_service.handle_notification(notification)
+        for notification in decode_server_notifications(request.data):
+            room_service.handle_notification(notification)
 
         return "Webhook Notification Received", 200

fishjam/__init__.py

@@ -12,7 +12,7 @@
 from fishjam._openapi_client.models import PeerMetadata
 
 # API
-from fishjam._webhook_notifier import receive_binary
+from fishjam._webhook_notifier import decode_server_notifications, receive_binary
 from fishjam._ws_notifier import FishjamNotifier
 from fishjam.api._fishjam_client import (
     AgentOptions,
@@ -31,6 +31,7 @@
 __all__ = [
     "FishjamClient",
     "FishjamNotifier",
+    "decode_server_notifications",
     "receive_binary",
     "PeerMetadata",
     "PeerOptions",

fishjam/_webhook_notifier.py

@@ -1,5 +1,6 @@
 """Module for decoding received webhook notifications from Fishjam."""
 
+import warnings
 from typing import List, Union
 
 import betterproto
@@ -40,11 +41,43 @@ def _unpack_batch(
     return notifications
 
 
+def decode_server_notifications(binary: bytes) -> List[AllowedNotification]:
+    """Decode a received protobuf payload into a list of notifications.
+
+    Handles both single notifications and batches transparently: a single
+    notification is returned as a one-element list, a batch is unpacked into
+    its members (in order), and anything unsupported yields an empty list.
+
+    The available notifications are listed in the `fishjam.events` module.
+
+    Args:
+        binary: The raw binary data received from the webhook.
+
+    Returns:
+        list[AllowedNotification]: The decoded notifications, in order. Empty
+            when the payload carries no supported notification.
+    """
+    message = ServerMessage().parse(binary)
+    _which, content = betterproto.which_one_of(message, "content")
+
+    if isinstance(content, ServerMessageNotificationBatch):
+        return _unpack_batch(content)
+
+    if isinstance(content, ALLOWED_NOTIFICATIONS):
+        return [content]
+
+    return []
+
+
 def receive_binary(
     binary: bytes,
 ) -> Union[AllowedNotification, List[AllowedNotification], None]:
     """Transforms a received protobuf notification into a notification instance.
 
+    .. deprecated::
+        Use `decode_server_notifications` instead, which always returns a list
+        and handles batched payloads with a single, consistent return type.
+
     The available notifications are listed in `fishjam.events` module.
 
     Args:
@@ -56,6 +89,12 @@ def receive_binary(
             payload is a batch (webhook batching enabled).
         None: When the payload is not a supported notification.
     """
+    warnings.warn(
+        "receive_binary is deprecated; use decode_server_notifications instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+
     message = ServerMessage().parse(binary)
     _which, content = betterproto.which_one_of(message, "content")
 

tests/support/webhook_notifier.py

@@ -2,7 +2,7 @@
 
 from flask import Flask, Response, request
 
-from fishjam import receive_binary
+from fishjam import decode_server_notifications
 
 app = Flask(__name__)
 QUEUES = None
@@ -16,12 +16,9 @@ def respond_default():
 @app.route("/webhook", methods=["POST"])
 def respond_root():
     data = request.get_data()
-    msg = receive_binary(data)
-    if msg is not None:
-        notifications = msg if isinstance(msg, list) else [msg]
-        for notification in notifications:
-            for q in QUEUES.values():
-                q.put(notification)
+    for notification in decode_server_notifications(data):
+        for q in QUEUES.values():
+            q.put(notification)
 
     return Response(status=200)
 

tests/test_allowed_notifications.py

@@ -20,8 +20,8 @@
     "ServerMessageTrackForwarding",
     "ServerMessageTrackForwardingRemoved",
     "ServerMessageVadNotification",
-    # Transport wrapper, not a user-facing event: `receive_binary` unpacks it
-    # into the individual notifications it carries.
+    # Transport wrapper, not a user-facing event: `decode_server_notifications`
+    # unpacks it into the individual notifications it carries.
     "ServerMessageNotificationBatch",
     # Deprecated in the proto.
     "ServerMessageStreamConnected",

tests/test_webhook_notifier.py

@@ -1,4 +1,6 @@
-from fishjam import receive_binary
+import pytest
+
+from fishjam import decode_server_notifications, receive_binary
 from fishjam.events import (
     ServerMessagePeerConnected,
     ServerMessageRoomCreated,
@@ -87,3 +89,85 @@ def test_empty_batch_returns_empty_list():
     result = receive_binary(binary)
 
     assert result == []
+
+
+def test_receive_binary_is_deprecated():
+    binary = bytes(ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1")))
+
+    with pytest.warns(DeprecationWarning):
+        receive_binary(binary)
+
+
+def test_decode_single_notification_returns_one_element_list():
+    binary = bytes(ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1")))
+
+    result = decode_server_notifications(binary)
+
+    assert isinstance(result, list)
+    assert [type(n) for n in result] == [ServerMessageRoomCreated]
+    assert result[0].room_id == "r1"
+
+
+def test_decode_unsupported_single_message_returns_empty_list():
+    binary = bytes(ServerMessage(auth_request=ServerMessageAuthRequest(token="t")))
+
+    assert decode_server_notifications(binary) == []
+
+
+def test_decode_batch_is_unpacked_into_ordered_list():
+    binary = bytes(
+        ServerMessage(
+            notification_batch=ServerMessageNotificationBatch(
+                notifications=[
+                    ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1")),
+                    ServerMessage(
+                        peer_connected=ServerMessagePeerConnected(
+                            room_id="r1", peer_id="p1"
+                        )
+                    ),
+                    ServerMessage(room_deleted=ServerMessageRoomDeleted(room_id="r1")),
+                ]
+            )
+        )
+    )
+
+    result = decode_server_notifications(binary)
+
+    assert [type(n) for n in result] == [
+        ServerMessageRoomCreated,
+        ServerMessagePeerConnected,
+        ServerMessageRoomDeleted,
+    ]
+    assert result[0].room_id == "r1"
+    assert result[1].peer_id == "p1"
+
+
+def test_decode_batch_filters_out_unsupported_members():
+    binary = bytes(
+        ServerMessage(
+            notification_batch=ServerMessageNotificationBatch(
+                notifications=[
+                    ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1")),
+                    ServerMessage(auth_request=ServerMessageAuthRequest(token="t")),
+                    ServerMessage(room_deleted=ServerMessageRoomDeleted(room_id="r1")),
+                ]
+            )
+        )
+    )
+
+    result = decode_server_notifications(binary)
+
+    assert [type(n) for n in result] == [
+        ServerMessageRoomCreated,
+        ServerMessageRoomDeleted,
+    ]
+
+
+def test_decode_empty_batch_returns_empty_list():
+    binary = bytes(
+        ServerMessage(
+            notification_batch=ServerMessageNotificationBatch(notifications=[])
+        )
+    )
+
+    assert decode_server_notifications(binary) == []