Mule4客户端连接Python WebSocket服务器无法接收回复的配置问题

我搭建了一个Python WebSocket服务器，通过Mule4客户端连接后，能成功发送消息到服务器，但收不到服务器的回复。Python服务器日志显示已经执行websocket.send(reply)发送了回复，我知道需要配置相关组件来接收消息，但不清楚如何正确配置“On outbound message”或“On inbound message”来捕获并记录这个回复。

Python服务器代码 #

import asyncio
import websockets

# create handler for each connection
async def handler(websocket, path):
    print("client just connnected")
    data = await websocket.recv()
    print("data:" + data)
    reply = f"Data recieved as:  {data}!"
    await websocket.send(reply)
    print("await completed")

start_server = websockets.serve(handler, "localhost", 8000)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

Mule4客户端当前配置 #

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:file="http://www.mulesoft.org/schema/mule/file" xmlns:sockets="http://www.mulesoft.org/schema/mule/sockets"
    xmlns:os="http://www.mulesoft.org/schema/mule/os"
    xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core" xmlns:websocket="http://www.mulesoft.org/schema/mule/websocket" xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/websocket http://www.mulesoft.org/schema/mule/websocket/current/mule-websocket.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
http://www.mulesoft.org/schema/mule/os http://www.mulesoft.org/schema/mule/os/current/mule-os.xsd
http://www.mulesoft.org/schema/mule/sockets http://www.mulesoft.org/schema/mule/sockets/current/mule-sockets.xsd
http://www.mulesoft.org/schema/mule/file http://www.mulesoft.org/schema/mule/file/current/mule-file.xsd">
    <http:listener-config name="HTTP_Listener_config" doc:name="HTTP Listener config" doc:id="71f64084-8631-411b-ac3d-72753e0420dc" >
        <http:listener-connection host="0.0.0.0" port="8081" />
    </http:listener-config>
    <websocket:config name="WebSockets_Config" doc:name="WebSockets Config" doc:id="8b7e2026-1a43-40a5-b6f6-57f1e810e1e9" >
        <websocket:connection >
            <websocket:client-settings host="localhost" port="8000"/>
        </websocket:connection>
    </websocket:config>
    <websocket:config name="WebSockets_serve_config" doc:name="WebSockets Config" doc:id="6bb1f71b-fcb1-4bc7-bbbc-b3167d952092" >
        <websocket:connection >
            <websocket:server-settings listenerConfig="HTTP_Listener_config" />
        </websocket:connection>
    </websocket:config>
    <flow name="scorewsclientFlow" doc:id="44652992-9b94-45e2-9942-3e965fdd4fd2" >
        <http:listener doc:name="Listener" doc:id="76ea16ed-0f8d-4fe0-996d-750b65d81af3" config-ref="HTTP_Listener_config" path="/connect"/>
        <set-variable value='#[correlationId ++ "-" ++ attributes.queryParams.client_name]' doc:name="Set Variable" doc:id="b9f3920f-1244-465b-aaeb-820596206708" variableName="socketid"/>
        <os:store doc:name="Store" doc:id="8143c9a5-af90-46b8-b158-ba2cfed6fce9" key="#[attributes.queryParams.client_name]">
            <os:value ><![CDATA[#[vars.socketid]]]></os:value>
        </os:store>
        <websocket:open-outbound-socket doc:name="Open outbound socket" doc:id="f0d6d235-2c74-4546-921e-8fe9e12c084a" config-ref="WebSockets_Config" path="/open" socketId="#[vars.socketid]"/>
        <ee:transform doc:name="Transform Message" doc:id="93d5e10f-babf-45bf-93f5-bf2c8e047bf1">
            <ee:message>
                <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    message: "Connection successful"
}]]></ee:set-payload>
            </ee:message>
        </ee:transform>
    </flow>
    <flow name="scorewsclientFlow1" doc:id="690325a5-c162-49f8-b056-e02b0150e999" >
        <http:listener doc:name="Listener" doc:id="6496973b-be0e-4a94-96c6-803951688f5f" config-ref="HTTP_Listener_config" path="/send"/>
        <set-variable value="#[attributes.queryParams.client_name]" doc:name="Set Variable" doc:id="3b6e1023-a9b9-49fa-a11d-67c54829246b" variableName="client_name" />
        <os:retrieve doc:name="Retrieve" doc:id="7727bec0-8258-4b18-934f-269d9cbd1f1c" key="#[vars.client_name]"/>
        <set-variable value="#[payload]" doc:name="Set Variable" doc:id="6ab9d0d9-9d67-4af6-8a87-b2d5aa8eed04" variableName="socketid"/>
        <ee:transform doc:name="Message-subscription" doc:id="68c440ae-b88a-476a-a696-393caa79b063">
            <ee:message>
                <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    client_name: vars.client_name,
    match_id: "12345"
}]]></ee:set-payload>
            </ee:message>
        </ee:transform>
        <websocket:send doc:name="Send" doc:id="5f0ba317-3025-421b-9837-dd5bddeabdc7" config-ref="WebSockets_Config" socketId="#[vars.socketid]">
        </websocket:send>
        <ee:transform doc:name="Transform Message" doc:id="cde387c5-cd58-4f78-9a41-49076be68fff">
            <ee:message>
                <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    socketid: vars.socketid,
}]]></ee:set-payload>
            </ee:message>
        </ee:transform>
    </flow>
    <flow name="scorewsclientFlow3" doc:id="42b0a800-a027-41f7-a459-3958040eed1a" >
        <websocket:inbound-listener doc:name="On New Inbound Message" doc:id="706960f7-3f04-48a2-b95c-a66f9c62edd6" config-ref="WebSockets_serve_config" path="/"/>
        <logger level="INFO" doc:name="Logger" doc:id="5c99e3ae-f829-4f35-ac70-239002ea32f0" message="#[payload]"/>
    </flow>
</mule>

解决方案 #

你当前的scorewsclientFlow3用了websocket:inbound-listener，但这个组件是用来做WebSocket服务器接收外部连接的，和你作为客户端接收服务器回复的需求不匹配。正确的做法是使用websocket:on-outbound-message组件，专门监听客户端出站WebSocket连接收到的服务器回复。

具体修改步骤 #

移除或修改原有的scorewsclientFlow3，添加一个新的流来处理回复：

<flow name="ws-client-reply-handler" doc:id="a1b2c3d4-5678-90ef-ghij-klmnopqrstuv">
    <!-- 监听客户端出站套接字的回复，绑定客户端WebSocket配置 -->
    <websocket:on-outbound-message doc:name="On Outbound Message" doc:id="wxyz-1234-5678-abcd-efghijklmnop" 
        config-ref="WebSockets_Config" socketId="#[vars.socketid]">
    </websocket:on-outbound-message>
    <!-- 记录服务器回复内容 -->
    <logger level="INFO" doc:name="Logger" doc:id="1234-abcd-5678-wxyz-90abcdefghij" 
        message="收到Python服务器回复：#[payload]"/>
</flow>

关键注意点 #

• websocket:on-outbound-message是客户端接收服务器回复的专用组件，必须绑定你的客户端WebSocket配置（即WebSockets_Config），而不是服务器端的配置。
• 如果需要监听所有客户端出站套接字的回复，可以去掉socketId属性，这样就能捕获所有通过该配置建立的连接的回复。
• 你的Python服务器发送的是字符串格式消息，Mule可以直接解析，无需额外转换。

内容的提问来源于stack exchange，提问作者wout