c++ - 与 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