与 Mule/ActiveMQ 和 C++ Stomp 的客户端通信

Question

我有一个向队列发送消息的外部 C++ STOMP 客户端

myqueue

并订阅一个主题

mytopic

配置接收消息的流程（eclipse mars中的Mule插件），修改消息并传输回显响应：

<jms:activemq-connector name="Active_MQ" specification="1.1" brokerURL="tcp://localhost:61616" validateConnections="true" doc:name="Active MQ"/>

  <http:listener-config name="HTTP" host="localhost" port="8081"  doc:name="HTTP Listener Configuration"/>

   <flow name="jmsFlow">

        <jms:inbound-endpoint  queue="myqueue" connector-ref="Active_MQ" doc:name="JMS">

            <jms:transaction action="NONE"/>

        </jms:inbound-endpoint>

        <logger message="#[string: Logger1 Response: #[payload]]" level="INFO" doc:name="Logger1"/>

        <response>

            <echo-component doc:name="Echo"/>

        </response>

        <component class="org.mule.java.JavaClient" doc:name="Java"/>

    </flow>

C++ STOMP 客户端：

static BoostStomp*  stomp_client;
static string       notifications_topic = "mytopic";
static string       registration_queue = "myqueue";
static string       result("");
static int          number = 0;

bool subscription_callback(STOMP::Frame& _frame)
{
    number = _frame.body().v.size();
    result = _frame.body().c_str();
    return(true);
}

//int main(int argc, char *argv[])
int main(int argc, char *argv[])
{
    string  stomp_host = "localhost";
    int     stomp_port = 61613;

    char* msg = argv[0];
    int nmbr = 1;
    char* mymsg = new char[nmbr];
    strncpy(mymsg,msg,nmbr);
    mymsg[nmbr] = '\0';
    string msgstr = string(mymsg);
    try
    {
        // initiate a new BoostStomp client
        stomp_client = new BoostStomp(stomp_host, stomp_port);

        // start the client, (by connecting to the STOMP server)
        stomp_client->start();//(user, pass);

        // subscribe to a channel
        stomp_client->subscribe(notifications_topic, (STOMP::pfnOnStompMessage_t) &subscription_callback);

        // construct a headermap
        STOMP::hdrmap headers;
        string body = string("mymessage");

        // add an outgoing message to the queue
        stomp_client->send(registration_queue, headers, body);

        Sleep(10000);
        nmbr = number;
        strncpy(msg,result.c_str(),nmbr);
        msg[nmbr] = '\0';
        cout << "return message is " << result.c_str();
        result.clear();

        Sleep(10000);
        stomp_client->unsubscribe(notifications_topic);
        Sleep(1000);
        stomp_client->stop();
        delete stomp_client;
    }
    catch (std::exception& e)
    {
        cerr << "Error in BoostStomp: " << e.what() << "\n";
        return 1;
    }
    cout << "Call of test works!" << endl;
    return 0;
}

我不确定如何将修改后的消息发送到主题

mytopic

而不是回声响应。有什么建议吗？

也许，另一种选择是使用实现 STOMP 消息发送的 Java 应用：

StompConnection connection = new StompConnection();
connection.open("localhost", 61613);
connection.connect("","");

connection.send("/mytopic", msg.toString());

接收消息

connection.subscribe("/mytopic", Subscribe.AckModeValues.CLIENT);
StompFrame frame = connection.receive();
System.out.println("JavaClient received message: " + frame.getBody());
connection.disconnect();

有效，但外部 STOMP 客户端仍然没有收到它。 STOMP cout 是：

[12:03:28: 00625D48] BoostStomp:starting...
[12:03:28: 00625D48] BoostStomp:STOMP: Connecting to [::1]:61613...
[12:03:28: 00625D48] BoostStomp:STOMP TCP connection to [::1]:61613 is active
[12:03:28: 00625D48] BoostStomp:Sending CONNECT frame...
[12:03:28: 0062E698] BoostStomp:Worker thread: starting...
[12:03:28: 0062E698] BoostStomp:server supports STOMP version 1.1
waiting for answer
[12:03:29: 0062E698] BoostStomp:Sending SUBSCRIBE frame...
[12:03:29: 0062E698] BoostStomp:Sent!
[12:03:29: 0062E698] BoostStomp:Sending SEND frame...
[12:03:29: 0062E698] BoostStomp:Sent!

Answer 1

下面的截图是关于如何使用 JMS(ActiveMQ) 主题向订阅者发送消息的示例。

Active MQ send topic

Answer 2

嗨@LeBro，请参阅下面的示例源代码示例：

    <?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:jms="http://www.mulesoft.org/schema/mule/jms" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
    xmlns:spring="http://www.springframework.org/schema/beans" 
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/current/mule-jms.xsd">
    <jms:activemq-connector name="Active_MQ" brokerURL="tcp://localhost:61616" validateConnections="true" doc:name="Active MQ"/>
    <jms:activemq-connector name="Active_MQ_Outbound" brokerURL="tcp://localhost:61616" validateConnections="true" doc:name="Active MQ"/>
    <flow name="active-mq-testFlow">
        <jms:inbound-endpoint queue="myqueue" connector-ref="Active_MQ" doc:name="JMS"/>
        <logger level="INFO" doc:name="Logger"/>
        <set-payload value="#['Transformed Message : ' + payload]" doc:name="Set Payload"/>
        <jms:outbound-endpoint topic="mytopic" connector-ref="Active_MQ_Outbound" doc:name="JMS Outbound"/>
    </flow>
</mule>

尝试使用此 url 访问您的本地 ActiveMQ： http://localhost:8161/

然后尝试运行 mule 应用程序。

mule 应用成功消费消息，然后将其发送到主题。