feat: automatic routing of message for slack example using discriminators (#1814)

Co-authored-by: Adi-204 <adiboghawala@gmail.com>
Co-authored-by: Lukasz Gornicki <lpgornicki@gmail.com>

Author: Adi-204

.changeset/slack-auto-routing.md

@@ -0,0 +1,58 @@
+---
+"@asyncapi/generator": minor
+"@asyncapi/generator-components": minor
+"@asyncapi/generator-helpers": minor
+---
+
+- **Updated Component**: `OnMessage` (Python) - Added discriminator-based routing logic that automatically dispatches messages to operation-specific handlers before falling back to generic handlers
+- **New Helpers**: 
+  - `getMessageDiscriminatorData` - Extracts discriminator key and value from individual messages
+  - `getMessageDiscriminatorsFromOperations` - Collects all discriminator metadata from receive operations
+- Enhanced Python webSocket client generation with **automatic operation-based message routing**:
+
+## How python routing works
+
+- Generated WebSocket clients now automatically route incoming messages to operation-specific handlers based on message discriminators. Users can register handlers for specific message types without manually parsing or filtering messages.
+- When a message arrives, the client checks it against registered discriminators (e.g., `type: "hello"`, `type: "events_api"`)
+- If a match is found, the message is routed to the specific operation handler (e.g., `onHelloMessage`, `onEvent`)
+- If no match is found, the message falls back to generic message handlers
+- This enables clean separation of message handling logic based on message types
+
+> `discriminator` is a `string` field that you can add to any AsyncAPI Schema. This also means that it is limited to AsyncAPI Schema only, and it won't work with other schema formats, like for example, Avro. 
+
+The implementation automatically derives discriminator information from your AsyncAPI document:
+- Discriminator `key` is extracted from the `discriminator` field in your AsyncAPI spec
+- Discriminator `value` is extracted from the `const` property defined in message schemas
+
+Example AsyncAPI Schema with `discriminator` and `const`:
+```yaml
+schemas:
+    hello:
+      type: object
+      discriminator: type # you specify name of property
+      properties:
+        type:
+          type: string
+          const: hello # you specify the value of the discriminator property that is used for routing
+          description: A hello string confirming WebSocket connection
+```
+
+## Fallback
+
+When defaults aren't available in the AsyncAPI document, users must provide **both** `discriminator_key` and `discriminator_value` when registering handlers. Providing only one parameter is not supported - you must provide either both or neither.
+
+> **Why this limitation exists**: When a receive operation has multiple messages sharing the same discriminator key (e.g., all use `"type"` field), we need the specific value (e.g., `"hello"`, `"disconnect"`) to distinguish between them. Without both pieces of information, the routing becomes ambiguous.
+
+Example:
+
+```python
+# Default case - discriminator info auto-derived from AsyncAPI doc
+client.register_on_hello_message_handler(my_handler)
+
+# Custom case - must provide both key AND value
+client.register_on_hello_message_handler(
+    my_handler,
+    discriminator_key="message_type",
+    discriminator_value="custom_hello"
+)
+```

packages/components/src/components/OnMessage.js

@@ -32,7 +32,33 @@ const websocketOnMessageMethod = {
   python: () => {
     return {
       onMessageMethod: `def on_message(self, ws, message):
-  self.handle_message(message)`
+  # Parse message for routing
+  try:
+      parsed_message = json.loads(message)
+  except:
+      parsed_message = message
+
+  handled = False
+
+  # Check each operation's discriminator
+  for discriminator in self.receive_operation_discriminators:
+    key = discriminator.get("key")
+    value = discriminator.get("value")
+    operation_id = discriminator.get("operation_id")
+
+    # Check if message matches this discriminator
+    if key and isinstance(parsed_message, dict) and parsed_message.get(key) == value:
+      handler = self.receive_operation_handlers.get(operation_id)
+      if handler:
+        try:
+          handler(message)
+          handled = True
+        except Exception as error:
+          print(f"Error in {operation_id} handler: {error}")
+
+  # Fallback to generic handlers if not handled
+  if not handled:
+      self.handle_message(message)`
     };
   },
   dart: () => {

packages/components/src/index.js

@@ -17,4 +17,4 @@ export { CoreMethods } from './components/readme/CoreMethods';
 export { Installation } from './components/readme/Installation';
 export { Usage } from './components/readme/Usage';
 export { Overview } from './components/readme/Overview';
-export { Readme } from './components/readme/Readme';
+export { Readme } from './components/readme/Readme';

packages/components/test/components/__snapshots__/AvailableOperations.test.js.snap

@@ -157,5 +157,11 @@ client.mixed_message_examples(123)
 
 
 
-#### \`operation_with_snake_case(payload)\`"
+#### \`operation_with_snake_case(payload)\`
+
+
+
+
+#### \`receiveMarketUpdate(payload)\`
+Receive market update message"
 `;

packages/components/test/components/__snapshots__/Connect.test.js.snap

@@ -98,7 +98,33 @@ exports[`Testing of Connect function render python Connect method 1`] = `
     print(\\"Connected to HoppscotchEchoWebSocketClient server\\")
 
   def on_message(self, ws, message):
-    self.handle_message(message)
+    # Parse message for routing
+    try:
+        parsed_message = json.loads(message)
+    except:
+        parsed_message = message
+
+    handled = False
+
+    # Check each operation's discriminator
+    for discriminator in self.receive_operation_discriminators:
+      key = discriminator.get(\\"key\\")
+      value = discriminator.get(\\"value\\")
+      operation_id = discriminator.get(\\"operation_id\\")
+
+      # Check if message matches this discriminator
+      if key and isinstance(parsed_message, dict) and parsed_message.get(key) == value:
+        handler = self.receive_operation_handlers.get(operation_id)
+        if handler:
+          try:
+            handler(message)
+            handled = True
+          except Exception as error:
+            print(f\\"Error in {operation_id} handler: {error}\\")
+
+    # Fallback to generic handlers if not handled
+    if not handled:
+        self.handle_message(message)
 
   def on_error(self, ws, error):
     print(\\"WebSocket Error:\\", error)

packages/components/test/components/__snapshots__/OnMessage.test.js.snap

@@ -31,5 +31,31 @@ exports[`OnMessage renders per language renders javascript OnMessage method 1`]
 
 exports[`OnMessage renders per language renders python OnMessage method 1`] = `
 "def on_message(self, ws, message):
-  self.handle_message(message)"
+  # Parse message for routing
+  try:
+      parsed_message = json.loads(message)
+  except:
+      parsed_message = message
+
+  handled = False
+
+  # Check each operation's discriminator
+  for discriminator in self.receive_operation_discriminators:
+    key = discriminator.get(\\"key\\")
+    value = discriminator.get(\\"value\\")
+    operation_id = discriminator.get(\\"operation_id\\")
+
+    # Check if message matches this discriminator
+    if key and isinstance(parsed_message, dict) and parsed_message.get(key) == value:
+      handler = self.receive_operation_handlers.get(operation_id)
+      if handler:
+        try:
+          handler(message)
+          handled = True
+        except Exception as error:
+          print(f\\"Error in {operation_id} handler: {error}\\")
+
+  # Fallback to generic handlers if not handled
+  if not handled:
+      self.handle_message(message)"
 `;

packages/helpers/src/discriminators.js

@@ -0,0 +1,72 @@
+/**
+* Extracts discriminator metadata from a message for operation routing.
+* 
+* @param {object} message - The AsyncAPI message object containing payload and discriminator info.
+* @param {string} operationId - The operation ID associated with this message.
+* @returns {object|null} An object with key, value, and operation_id if discriminator is valid; otherwise null.
+*/
+const getMessageDiscriminatorData = (message, operationId) => {
+  const payload = message.payload();
+  if (!payload) {
+    return null;
+  }
+  const discriminator = payload.discriminator();
+
+  if (!discriminator) {
+    return null;
+  }
+
+  const discriminator_key = discriminator;
+  const properties = payload.properties();
+
+  if (!properties || !properties[discriminator_key]) {
+    return null;
+  }
+
+  const discriminatorProperty = properties[discriminator_key];
+  const discriminator_value = discriminatorProperty.const();
+
+  if (!discriminator_value) {
+    return null;
+  }
+
+  return {
+    key: discriminator_key,
+    value: discriminator_value,
+    operation_id: operationId
+  };
+};
+
+/**
+ * Get discriminator metadata from all messages across a list of AsyncAPI operations.
+ *
+ * @param {Array<object>} operations - List of AsyncAPI Operation objects.
+ * @returns {Array<object>} Array of discriminator metadata objects with key, value, and operation_id.
+ */
+const getMessageDiscriminatorsFromOperations = (operations) => {
+  if (!operations || !Array.isArray(operations)) {
+    return [];
+  }
+  const operationDiscriminators = [];
+
+  operations.forEach((operation) => {
+    const operationId = operation.id();
+    const messages = operation.messages().all();
+
+    messages
+      .filter(message => message.hasPayload())
+      .forEach(message => {
+        const discriminatorData = getMessageDiscriminatorData(message, operationId);
+        if (discriminatorData) {
+          operationDiscriminators.push(discriminatorData);
+        }
+      });
+  });
+
+  return operationDiscriminators;
+};
+
+module.exports = {
+  getMessageDiscriminatorData,
+  getMessageDiscriminatorsFromOperations
+};

packages/helpers/src/index.js

@@ -1,6 +1,7 @@
 const { getMessageExamples, getOperationMessages } = require('./operations');
 const { getServerUrl, getServer, getServerHost, getServerProtocol }  = require('./servers');
-const { getClientName, getInfo, toSnakeCase, toCamelCase, getTitle, lowerFirst, upperFirst} = require('./utils');
+const { getClientName, getInfo, toSnakeCase, toCamelCase, getTitle, lowerFirst, upperFirst } = require('./utils');
+const { getMessageDiscriminatorData, getMessageDiscriminatorsFromOperations } = require('./discriminators');
 const { getQueryParams } = require('./bindings');
 const { cleanTestResultPaths, verifyDirectoryStructure, getDirElementsRecursive, buildParams, listFiles, hasNestedConfig} = require('./testing');
 const { JavaModelsPresets } = require('./ModelsPresets');
@@ -26,5 +27,7 @@ module.exports = {
   getDirElementsRecursive,
   buildParams,
   hasNestedConfig,
-  JavaModelsPresets
+  JavaModelsPresets,
+  getMessageDiscriminatorData, 
+  getMessageDiscriminatorsFromOperations
 };

packages/helpers/test/__fixtures__/asyncapi-websocket-query.yml

@@ -74,6 +74,36 @@ channels:
             payload: false
           - name: number
             payload: 123
+      MarketUpdateMessage:
+        summary: Market update message with discriminator
+        payload:
+          type: object
+          discriminator: messageType
+          properties:
+            messageType:
+              type: string
+              const: marketUpdate
+            data:
+              type: string
+      MessageWithoutDiscriminator:
+        summary: Message without discriminator
+        payload:
+          type: object
+          properties:
+            messageType:
+              type: string
+            data:
+              type: string
+      MessageWithDiscriminatorNoConst:
+        summary: Message with discriminator but no const value
+        payload:
+          type: object
+          discriminator: messageType
+          properties:
+            messageType:
+              type: string
+            data:
+              type: string
   wsBindingNoQuery:
     address: '/no-query'
     bindings:
@@ -163,6 +193,16 @@ operations:
       $ref: '#/channels/marketDataV1'
     messages:
       - $ref: '#/channels/marketDataV1/messages/noExamples'
+  
+  receiveMarketUpdate:
+    action: receive
+    channel:
+      $ref: '#/channels/marketDataV1'
+    summary: Receive market update message
+    messages:
+      - $ref: '#/channels/marketDataV1/messages/MarketUpdateMessage'
+      - $ref: '#/channels/marketDataV1/messages/MessageWithoutDiscriminator'
+      - $ref: '#/channels/marketDataV1/messages/MessageWithDiscriminatorNoConst'
 
 components:
   parameters: