WIP

Author: huynguyengl99

CHANGELOG.md

@@ -88,7 +88,7 @@
 ### Feat
 
 - **python310**: add support for python version 310
-- **websocket**: add kind to send_group_message to handle both pydantic message and json case
+- **websocket**: add kind to broadcast_message to handle both pydantic message and json case
 
 ### Fix
 

README.rst

@@ -113,9 +113,9 @@ You can use these parameters in different combinations:
     # With group messaging
     class GroupConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
         async def receive_message(self, message: ChatMessage, **kwargs: Any) -> None:
-            # Send typed group messages using send_group_message
+            # Send typed group messages using broadcast_message
             group_msg = MemberMessage(payload={"content": "Hello group!"})
-            await self.send_group_message(group_msg)
+            await self.broadcast_message(group_msg)
 
     # Complete example with all generic parameters
     class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage, ChatEvent, Room]):

chanx/asyncapi/generator.py

@@ -135,23 +135,22 @@ def build_operations(self) -> None:
         for route in self.routes:
             consumer = route.consumer
 
-            for action, handler_info in consumer._MESSAGE_HANDLER_INFO_MAP.items():
+            for handler_info in consumer._MESSAGE_HANDLER_INFO_MAP.values():
                 self._build_single_operation(
-                    action, consumer, route, handler_info, is_event=False
+                    handler_info, consumer, route, is_event=False
                 )
 
             # Build operations from event handlers (send operations)
             for action, handler_info in consumer._EVENT_HANDLER_INFO_MAP.items():
                 self._build_single_operation(
-                    action, consumer, route, handler_info, is_event=True
+                    handler_info, consumer, route, is_event=True
                 )
 
     def _build_single_operation(
         self,
-        action: str,
+        handler_info: AsyncAPIHandlerInfo,
         consumer: AsyncJsonWebsocketConsumer,
         route: RouteInfo,
-        handler_info: AsyncAPIHandlerInfo,
         is_event=False,
     ) -> None:
         """
@@ -164,9 +163,9 @@ def _build_single_operation(
         Returns:
             AsyncAPI operation definition.
         """
-        action_name = action
-        if action in self._operation_names:
-            action_name = "_".join((consumer.snake_name, action))
+        action_name = handler_info["action"]
+        if action_name in self._operation_names:
+            action_name = "_".join((consumer.snake_name, action_name))
 
         channel_name = self._route_channel_mapping[route.path]
         operation: dict[str, Any] = {

chanx/core/adapter.py

@@ -2,19 +2,18 @@
 
 IS_USING_DJANGO = os.environ.get("DJANGO_SETTINGS_MODULE", False)
 
-if IS_USING_DJANGO:
-    from channels.generic.websocket import AsyncJsonWebsocketConsumer
-    from channels.layers import get_channel_layer
-    from channels.testing import WebsocketCommunicator
-
-    from asgiref.typing import WebSocketConnectEvent, WebSocketDisconnectEvent
-
-else:
-    from fast_channels.consumer import AsyncJsonWebsocketConsumer
-    from fast_channels.layers import get_channel_layer
-    from fast_channels.testing import WebsocketCommunicator
-    from fast_channels.type_defs import WebSocketConnectEvent, WebSocketDisconnectEvent
-
+# if IS_USING_DJANGO:
+#     from channels.generic.websocket import AsyncJsonWebsocketConsumer
+#     from channels.layers import get_channel_layer
+#     from channels.testing import WebsocketCommunicator
+#
+#     from asgiref.typing import WebSocketConnectEvent, WebSocketDisconnectEvent
+#
+# else:
+from fast_channels.consumer import AsyncJsonWebsocketConsumer
+from fast_channels.layers import get_channel_layer
+from fast_channels.testing import WebsocketCommunicator
+from fast_channels.type_defs import WebSocketConnectEvent, WebSocketDisconnectEvent
 
 __all__ = [
     "get_channel_layer",

chanx/core/decorators.py

@@ -19,6 +19,7 @@
 
 _P = ParamSpec("_P")
 _R = TypeVar("_R")
+_T = TypeVar("_T")
 
 
 @overload
@@ -118,6 +119,7 @@ def decorator(func: Callable) -> Callable:  # noqa C901
         consumer_name = func.__qualname__.split(".")[0]
         handler_info: AsyncAPIHandlerInfo = {
             "action": final_action,
+            "message_action": final_input_type.model_fields["action"].default,
             "input_type": final_input_type,
             "output_type": final_output_type,
             "method_name": func.__name__,
@@ -291,7 +293,7 @@ def channel(
     name: str | None = None,
     description: str | None = None,
     tags: list[str] | None = None,
-) -> Callable[[type], type]:
+) -> Callable[[_T], _T]:
     """
     Decorator for WebSocket consumer classes to add AsyncAPI channel metadata.
 
@@ -311,7 +313,7 @@ class AssistantConsumer(AsyncJsonWebsocketConsumer):
             pass
     """
 
-    def decorator(cls: type) -> type:
+    def decorator(cls: _T) -> _T:
         # Store channel information on the class
         channel_info: ChannelInfo = {
             "name": name,

chanx/core/websocket.py

@@ -106,13 +106,17 @@ def _process_handlers(cls) -> None:
 
             # Process WebSocket handlers
             if hasattr(attr, "_ws_handler_info"):
-                handler_info = attr._ws_handler_info
-                cls._MESSAGE_HANDLER_INFO_MAP[handler_info["action"]] = handler_info
+                handler_info: AsyncAPIHandlerInfo = attr._ws_handler_info
+                cls._MESSAGE_HANDLER_INFO_MAP[handler_info["message_action"]] = (
+                    handler_info
+                )
 
             # Process event handlers
             if hasattr(attr, "_event_handler_info"):
-                handler_info = attr._event_handler_info
-                cls._EVENT_HANDLER_INFO_MAP[handler_info["action"]] = handler_info
+                handler_info: AsyncAPIHandlerInfo = attr._event_handler_info
+                cls._EVENT_HANDLER_INFO_MAP[handler_info["message_action"]] = (
+                    handler_info
+                )
 
         # Extract types and build unions/adapters
         cls._build_adapters()
@@ -264,14 +268,23 @@ async def websocket_connect(self, message: WebSocketConnectEvent) -> None:
         await self.accept()
 
         # Authenticate the connection
-        auth_result = None
         if self.authenticator:
             auth_result = await self.authenticator.authenticate(self.scope)
 
             if not auth_result:
                 await self.close()
                 return
 
+        try:
+            for group in self.groups:
+                channel_layer = get_channel_layer(self.channel_layer_alias)
+                if channel_layer:
+                    await channel_layer.group_add(group, self.channel_name)
+        except AttributeError:
+            raise ValueError(
+                "BACKEND is unconfigured or doesn't support groups"
+            ) from None
+
         await self.post_authentication()
 
     async def websocket_disconnect(self, message: WebSocketDisconnectEvent) -> None:
@@ -367,10 +380,10 @@ async def receive_message(self, message: BaseMessage) -> None:
             message: The validated message object (BaseMessage instance)
         """
         # Extract the action from the discriminator field
-        action = getattr(message, self.discriminator_field)
+        message_action = getattr(message, self.discriminator_field)
 
-        # Find the handler for this action
-        handler_info = self.__class__._MESSAGE_HANDLER_INFO_MAP.get(action)
+        # Find the handler for this message_action
+        handler_info = self.__class__._MESSAGE_HANDLER_INFO_MAP.get(message_action)
 
         # Get the handler method by name
         method_name = handler_info["method_name"]
@@ -385,7 +398,7 @@ async def receive_message(self, message: BaseMessage) -> None:
                 await self.handle_result(result)
 
         except Exception as e:
-            await self.handle_error(e, action, message)
+            await self.handle_error(e, message_action, message)
 
     async def handle_result(self, result: BaseMessage) -> None:
         """
@@ -458,12 +471,12 @@ async def handle_error(
         await logger.aexception(f"Handler error for action '{action}': {error}")
 
     # Group operations methods
-    async def send_group_message(
+    async def broadcast_message(
         self,
         message: BaseMessage | dict[str, Any],
         groups: list[str] | None = None,
         *,
-        exclude_current: bool = True,
+        exclude_current: bool = False,
     ) -> None:
         """
         Send a BaseMessage object to one or more channel groups.
@@ -479,24 +492,27 @@ async def send_group_message(
             exclude_current: Whether to exclude the sending consumer from receiving
                             the broadcast (prevents echo effects)
         """
+        channel_layer = get_channel_layer(self.channel_layer_alias)
+        assert channel_layer
+
         if groups is None:
             groups = self.groups or []
 
         if isinstance(message, BaseMessage):
             message = message.model_dump(mode="json")
 
         for group in groups:
-            await self.channel_layer.group_send(
+            await channel_layer.group_send(
                 group,
                 {
-                    "type": "send_group_member",
+                    "type": "handle_group_message",
                     "message": message,
                     "exclude_current": exclude_current,
                     "from_channel": self.channel_name,
                 },
             )
 
-    async def send_group_member(self, event: BaseMessage) -> None:
+    async def handle_group_message(self, event: BaseMessage) -> None:
         """
         Handle incoming group message and relay to client.
 
@@ -534,7 +550,7 @@ async def send_group_member(self, event: BaseMessage) -> None:
 
     # Channel event system methods
     @classmethod
-    async def asend_event(
+    async def send_event(
         cls,
         channel_name: str,
         event: ReceiveEvent,
@@ -562,13 +578,13 @@ async def asend_event(
         )
 
     @classmethod
-    def send_event(
+    def send_event_sync(
         cls,
         channel_name: str,
         event: ReceiveEvent,
     ) -> None:
         """
-        Synchronous version of asend_event for use in Django tasks/views.
+        Synchronous version of send_event for use in Django tasks/views.
 
         This method provides the same functionality as asend_channel_event but
         can be called from synchronous code like Django tasks, views, or signals.
@@ -577,13 +593,13 @@ def send_event(
             channel_name: Channel name to send to
             event: The typed event to send (BaseMessage subclass)
         """
-        async_to_sync(cls.asend_event)(channel_name, event)
+        async_to_sync(cls.send_event)(channel_name, event)
 
     @classmethod
-    async def abroadcast_event(
+    async def broadcast_event(
         cls,
-        group_name: str,
         event: ReceiveEvent,
+        groups: Iterable[str] | None = None,
     ) -> None:
         """
         Broadcast a typed channel event to a channel group.
@@ -593,37 +609,41 @@ async def abroadcast_event(
         places where you don't have a consumer instance.
 
         Args:
-            group_name: Group name to broadcast the event to
             event: The typed event to broadcast (BaseMessage subclass)
+            groups: Groups to broadcast the event to
         """
         channel_layer = get_channel_layer(cls.channel_layer_alias)
         assert channel_layer is not None
 
-        await channel_layer.group_send(
-            group_name,
-            {
-                "type": "handle_channel_event",
-                "event_data": event.model_dump(mode="json"),
-            },
-        )
+        if groups is None:
+            groups = cls.groups or []
+
+        for group in groups:
+            await channel_layer.group_send(
+                group,
+                {
+                    "type": "handle_channel_event",
+                    "event_data": event.model_dump(mode="json"),
+                },
+            )
 
     @classmethod
-    def broadcast_event(
+    def broadcast_event_sync(
         cls,
-        group_name: str,
         event: ReceiveEvent,
+        groups: Iterable[str] | None = None,
     ) -> None:
         """
-        Synchronous version of abroadcast_event for use in Django tasks/views.
+        Synchronous version of broadcast_event for use in sync tasks/views.
 
-        This method provides the same functionality as abroadcast_event but
+        This method provides the same functionality as broadcast_event but
         can be called from synchronous code like Django tasks, views, or signals.
 
         Args:
-            group_name: Group name to broadcast to
             event: The typed event to broadcast (BaseMessage subclass)
+            groups: Groups to broadcast to
         """
-        async_to_sync(cls.abroadcast_event)(group_name, event)
+        async_to_sync(cls.broadcast_event)(event, groups)
 
     async def handle_channel_event(self, event_payload: EventPayload) -> None:
         """

chanx/type_defs.py

@@ -15,6 +15,7 @@ class HandlerInfo(TypedDict):
     """Information stored on handler functions by the @ws_handler decorator."""
 
     action: str
+    message_action: str
     input_type: type[BaseMessage] | None
     output_type: type[BaseMessage] | None
     method_name: str

docs/examples/chat.rst

@@ -142,7 +142,7 @@ Now we'll create our chat consumer with proper pattern matching:
             user = self.user
 
             # Send joined notification to the group
-            await self.send_group_message(
+            await self.broadcast_message(
                 UserJoinedPayload(
                     payload={
                         "username": user.username,
@@ -221,7 +221,7 @@ Now we'll create our chat consumer with proper pattern matching:
             chat_message = ChatMessagePayload(payload=text)
 
             # Broadcast to the group
-            await self.send_group_message(chat_message)
+            await self.broadcast_message(chat_message)
 
         async def save_message_to_db(self, user: User, room: ChatRoom, text: str) -> None:
             """Save chat message to database."""
@@ -247,7 +247,7 @@ Now we'll create our chat consumer with proper pattern matching:
                     room = self.obj
 
                     # Send user left notification
-                    await self.send_group_message(
+                    await self.broadcast_message(
                         UserLeftPayload(
                             payload={
                                 "username": user.username,

docs/introduction.rst

@@ -68,7 +68,7 @@ Chanx fills these gaps with a cohesive framework that provides:
       return [f"room_{self.room_id}"]
 
   async def receive_message(self, message: ChatMessage, **kwargs: Any) -> None:
-      await self.send_group_message(message)
+      await self.broadcast_message(message)
 
 4. **Enhanced URL Routing**: Django-style routing utilities for WebSocket endpoints with type hints support
 

docs/quick-start.rst

@@ -267,7 +267,7 @@ Now let's enhance our consumer to support group messaging. First, we need to add
                     username = getattr(self.user, 'username', 'Anonymous')
 
                     # Send to the whole group
-                    await self.send_group_message(
+                    await self.broadcast_message(
                         ChatGroupMessage(
                             payload=MessagePayload(content=f"{username}: {payload.content}")
                         )