| """ |
| Feature flags module for ComfyUI WebSocket protocol negotiation. |
| |
| This module handles capability negotiation between frontend and backend, |
| allowing graceful protocol evolution while maintaining backward compatibility. |
| """ |
|
|
| from typing import Any, Dict |
|
|
| from comfy.cli_args import args |
|
|
| |
| SERVER_FEATURE_FLAGS: Dict[str, Any] = { |
| "supports_preview_metadata": True, |
| "max_upload_size": args.max_upload_size * 1024 * 1024, |
| } |
|
|
|
|
| def get_connection_feature( |
| sockets_metadata: Dict[str, Dict[str, Any]], |
| sid: str, |
| feature_name: str, |
| default: Any = False |
| ) -> Any: |
| """ |
| Get a feature flag value for a specific connection. |
| |
| Args: |
| sockets_metadata: Dictionary of socket metadata |
| sid: Session ID of the connection |
| feature_name: Name of the feature to check |
| default: Default value if feature not found |
| |
| Returns: |
| Feature value or default if not found |
| """ |
| if sid not in sockets_metadata: |
| return default |
|
|
| return sockets_metadata[sid].get("feature_flags", {}).get(feature_name, default) |
|
|
|
|
| def supports_feature( |
| sockets_metadata: Dict[str, Dict[str, Any]], |
| sid: str, |
| feature_name: str |
| ) -> bool: |
| """ |
| Check if a connection supports a specific feature. |
| |
| Args: |
| sockets_metadata: Dictionary of socket metadata |
| sid: Session ID of the connection |
| feature_name: Name of the feature to check |
| |
| Returns: |
| Boolean indicating if feature is supported |
| """ |
| return get_connection_feature(sockets_metadata, sid, feature_name, False) is True |
|
|
|
|
| def get_server_features() -> Dict[str, Any]: |
| """ |
| Get the server's feature flags. |
| |
| Returns: |
| Dictionary of server feature flags |
| """ |
| return SERVER_FEATURE_FLAGS.copy() |
|
|