WSO2 EI 6.6.0 with RabbitMQ 3.8.5 - Unable to get exact posted payload in rabbitmq console

user_9907182 0
Viewed 151

I am just trying to post incoming payload to rabbitmq by using messageStore in WSO2 EI 6.6.0.

API-Code:

<api context="/testmqload" name="Test" xmlns="http://ws.apache.org/ns/synapse">
    <resource methods="POST">
        <inSequence>
            <log level="custom">
                <property expression="json-eval($)" name="Test"/>
            </log>
           
            <property name="OUT_ONLY" value="true"/>
            <property name="FORCE_SC_ACCEPTED" value="true" scope="axis2"/>
            <property name="messageType" value="application/json" scope="axis2"/>
            <store messageStore="SafaricomRequestStore"/>
        </inSequence>
        <outSequence/>
        <faultSequence/>
    </resource>
</api>

MessageStore:

<?xml version="1.0" encoding="UTF-8"?>
<messageStore class="org.apache.synapse.message.store.impl.rabbitmq.RabbitMQStore" name="SafaricomRequestStore" xmlns="http://ws.apache.org/ns/synapse">
    <parameter name="store.rabbitmq.host.name">puffin.rmq2.cloudamqp.com</parameter>
    <parameter name="store.producer.guaranteed.delivery.enable">false</parameter>
    <parameter name="store.rabbitmq.host.port">5672</parameter>
    <parameter name="store.rabbitmq.route.key">SafaricomRequestQueue</parameter>
    <parameter name="store.rabbitmq.username">username</parameter>
    <parameter name="store.rabbitmq.virtual.host">host</parameter>
    <parameter name="rabbitmq.connection.ssl.enabled">false</parameter>
    <parameter name="store.rabbitmq.exchange.name">amq.direct</parameter>
    <parameter name="store.rabbitmq.queue.name">SafaricomRequestQueueTest4</parameter>
    <parameter name="store.rabbitmq.password">password</parameter>
</messageStore>

Incoming Request:

{
    "result": {
        "message": {
            "M-PESA_Cash_Out": "10.00",
            "M-PESA_Fee": "15.27",
            "Transaction_Fee": "0.44",
            "acct_no": "4000000009",
            "date": "04-JAN-2022",
            "power_id": "cab48a3b620e4b3f8d7c2d13a9efced7",
            "receiving_mobile": "799999999",
            "time": "17:47:29",
            "total_amount": "25.71",
            "transaction_id": "1002201040000053",
            "wallet_bank": "M-PESA"
        },
        "new_txn_id": 1002201040000053,
        "status": "S"
    }
}

Message is posted in rabbitMQ like below

rabbitmq-screenshot

when i decoded above screenshot value by using base64 decoder online, getting below values.

decoded value

If you notice above screenshot, i am geting some other values in addition to posted payload which is marked as yellow above.

By using Inbound endpoint, i tried to fetch message from queue which will print message structure like below in sequence.

{
"text":"base64 encoded value...."
}

may i know why it is having other extra values too in addition with posted payload? or what can i do to fetch only posted payload from queue?

As per tmoasz suggested in comment, modified Code- which are mentioned below.

Modified API :

<?xml version="1.0" encoding="UTF-8"?>
<api context="/rabbitmqtest" name="RabbitMQTestAPI" xmlns="http://ws.apache.org/ns/synapse">
    <resource methods="POST">
        <inSequence>
         <log level="custom">
         <property name="RabbitMQTestAPI" value="is called***"/>
         <property name="IncomingRequest" expression="json-eval($)"/>
         </log>
            <property description="Initiate asynchronous mediation flow" name="OUT_ONLY" scope="default" type="STRING" value="true"/>
            <property description="Generate 202 response from mediation flow" name="FORCE_SC_ACCEPTED" scope="axis2" type="STRING" value="true"/>
            <send description="Publish request payload to AMQP endpoint">
                <endpoint key="RabbitMQ-QueueEP"/>
            </send>
        
        </inSequence>
        <outSequence/>
        <faultSequence/>
    </resource>
</api>

Endpoint Code:

<?xml version="1.0" encoding="UTF-8"?>
<endpoint name="RabbitMQ-QueueEP" xmlns="http://ws.apache.org/ns/synapse">
    <address uri="rabbitmq:/RabbitMQTestQueue?rabbitmq.server.host.name=puffin.rmq2.cloudamqp.com&amp;rabbitmq.server.port=5672&amp;rabbitmq.server.user.name=username&amp;rabbitmq.server.password=password&amp;rabbitmq.queue.name=RabbitMQTestQueue&amp;rabbitmq.exchange.name=amq.direct">
        <suspendOnFailure>
            <initialDuration>-1</initialDuration>
            <progressionFactor>-1</progressionFactor>
        </suspendOnFailure>
        <markForSuspension>
            <retriesBeforeSuspension>0</retriesBeforeSuspension>
        </markForSuspension>
    </address>
</endpoint>

Wire logs:

    [2022-01-06 19:01:11,282] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "POST /rabbitmqtest HTTP/1.1[\r][\n]"
[2022-01-06 19:01:11,283] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Content-Type: application/json[\r][\n]"
[2022-01-06 19:01:11,285] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "User-Agent: PostmanRuntime/7.28.4[\r][\n]"
[2022-01-06 19:01:11,286] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Accept: */*[\r][\n]"
[2022-01-06 19:01:11,289] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Cache-Control: no-cache[\r][\n]"
[2022-01-06 19:01:11,290] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Postman-Token: 75a4502d-7df0-444e-8ba8-39bcd6265204[\r][\n]"
[2022-01-06 19:01:11,290] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Host: localhost:8280[\r][\n]"
[2022-01-06 19:01:11,291] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Accept-Encoding: gzip, deflate, br[\r][\n]"
[2022-01-06 19:01:11,292] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Connection: keep-alive[\r][\n]"
[2022-01-06 19:01:11,292] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "Content-Length: 25[\r][\n]"
[2022-01-06 19:01:11,293] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "[\r][\n]"
[2022-01-06 19:01:11,300] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "{[\r][\n]"
[2022-01-06 19:01:11,301] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "    "name":"justin"[\r][\n]"
[2022-01-06 19:01:11,302] DEBUG {org.apache.synapse.transport.http.wire} - HTTP-Listener I/O dispatcher-4 >> "}"
[2022-01-06 19:01:11,306]  INFO {org.apache.synapse.mediators.builtin.LogMediator} - RabbitMQTestAPI = is called***, IncomingRequest = {
    "name":"justin"
}
[2022-01-06 19:01:11,309]  INFO {org.apache.axis2.transport.rabbitmq.RabbitMQConnectionFactory} - Initializing channel pool of 20
[2022-01-06 19:01:11,879] ERROR {org.apache.axis2.transport.rabbitmq.RabbitMQConnectionFactory} - [puffin.rmq2.cloudamqp.com_5672_username_passwordI_null_null_null_null_null_null_null_null_null] Error creating connection to RabbitMQ Broker. Reattempting to connect. java.io.IOException
        at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:105)
        at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:101)
        at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:123)
        at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:382)
        at com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory.newConnection(RecoveryAwareAMQConnectionFactory.java:58)
        at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.init(AutorecoveringConnection.java:103)
        at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:877)
        at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:839)
        at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:661)
        at org.apache.axis2.transport.rabbitmq.utils.RabbitMQUtils.createConnection(RabbitMQUtils.java:56)
        at org.apache.axis2.transport.rabbitmq.RabbitMQConnectionFactory.createConnection(RabbitMQConnectionFactory.java:159)
        at org.apache.axis2.transport.rabbitmq.RMQChannelPool.<init>(RMQChannelPool.java:20)
        at org.apache.axis2.transport.rabbitmq.RabbitMQConnectionFactory.initializeConnectionPool(RabbitMQConnectionFactory.java:392)
        at org.apache.axis2.transport.rabbitmq.RabbitMQConnectionFactoryManager.getConnectionFactory(RabbitMQConnectionFactoryManager.java:105)
        at org.apache.axis2.transport.rabbitmq.RabbitMQSender.getConnectionFactory(RabbitMQSender.java:135)
        at org.apache.axis2.transport.rabbitmq.RabbitMQSender.sendMessage(RabbitMQSender.java:81)
        at org.apache.axis2.transport.base.AbstractTransportSender.invoke(AbstractTransportSender.java:112)
        at org.apache.axis2.engine.AxisEngine.send(AxisEngine.java:442)
        at org.apache.axis2.description.OutOnlyAxisOperationClient.executeImpl(OutOnlyAxisOperation.java:297)
        at org.apache.axis2.client.OperationClient.execute(OperationClient.java:149)
        at org.apache.synapse.core.axis2.Axis2FlexibleMEPClient.send(Axis2FlexibleMEPClient.java:634)
        at org.apache.synapse.core.axis2.Axis2Sender.sendOn(Axis2Sender.java:85)
        at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.send(Axis2SynapseEnvironment.java:571)
        at org.apache.synapse.endpoints.AbstractEndpoint.send(AbstractEndpoint.java:408)
        at org.apache.synapse.endpoints.AddressEndpoint.send(AddressEndpoint.java:74)
        at org.apache.synapse.endpoints.IndirectEndpoint.send(IndirectEndpoint.java:56)
        at org.apache.synapse.mediators.builtin.SendMediator.mediate(SendMediator.java:123)
        at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:109)
        at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:71)
        at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158)
        at org.apache.synapse.rest.Resource.process(Resource.java:331)
        at org.apache.synapse.rest.API.process(API.java:441)
        at org.apache.synapse.rest.RESTRequestHandler.apiProcess(RESTRequestHandler.java:135)
        at org.apache.synapse.rest.RESTRequestHandler.dispatchToAPI(RESTRequestHandler.java:113)
        at org.apache.synapse.rest.RESTRequestHandler.process(RESTRequestHandler.java:71)
        at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.injectMessage(Axis2SynapseEnvironment.java:327)
        at org.apache.synapse.core.axis2.SynapseMessageReceiver.receive(SynapseMessageReceiver.java:98)
        at org.apache.axis2.engine.AxisEngine.receive(AxisEngine.java:180)
        at org.apache.synapse.transport.passthru.ServerWorker.processNonEntityEnclosingRESTHandler(ServerWorker.java:368)
        at org.apache.synapse.transport.passthru.ServerWorker.processEntityEnclosingRequest(ServerWorker.java:427)
        at org.apache.synapse.transport.passthru.ServerWorker.run(ServerWorker.java:182)
        at org.apache.axis2.transport.base.threads.NativeWorkerPool$1.run(NativeWorkerPool.java:172)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: com.rabbitmq.client.ShutdownSignalException: connection error; protocol method: #method<connection.close>(reply-code=530, reply-text=NOT_ALLOWED - access to vhost '/' refused for user 'username', class-id=10, method-id=40)
        at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66)
        at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:32)
        at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:366)
        at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:229)
        at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:117)
        ... 42 more
wso2-esb wso2-enterprise-integrator wso2 rabbitmq
2 Answers

If you want to send, just JSON message as payload to RabbitMQ, don't use message store. Message stores, are storing all synapse message context with payload for later processing - that is their purpose. So that is why you see in decoded RabbitMQ Payload not only the message, but also additional properties.

For sending JSON to RabbitMQ you can use just <send/> to a proper endpoint. Look as this documentation sample. I have also have made some sequance template for more easy use of sending messages to rabitmq, so next you can may look at this helpful post, and adapt to your own needs.

user_27605060

I have resolved above issue by making endpoint call with the use of send mediator as @tmoasz suggested instead of message store.

API Code:

<?xml version="1.0" encoding="UTF-8"?>
<api context="/rabbitmqtest" name="RabbitMQTestAPI" xmlns="http://ws.apache.org/ns/synapse">
    <resource methods="POST">
        <inSequence>
         <log level="custom">
         <property name="RabbitMQTestAPI" value="is called***"/>
         <property name="IncomingRequest" expression="json-eval($)"/>
         </log>
            <property description="Initiate asynchronous mediation flow" name="OUT_ONLY" scope="default" type="STRING" value="true"/>
            <property description="Generate 202 response from mediation flow" name="FORCE_SC_ACCEPTED" scope="axis2" type="STRING" value="true"/>
            <send description="Publish request payload to AMQP endpoint">
                <endpoint key="RabbitMQ-QueueEP"/>
            </send>
        
        </inSequence>
        <outSequence/>
        <faultSequence/>
    </resource>
</api>

RabbitMQ-QueueEP :

<?xml version="1.0" encoding="UTF-8"?>
<endpoint name="RabbitMQ-QueueEP" xmlns="http://ws.apache.org/ns/synapse">
    <address uri="rabbitmq:/AMQPConnectionFactory?rabbitmq.server.host.name=puffin.rmq2.cloudamqp.com&amp;rabbitmq.server.port=5672&amp;rabbitmq.server.user.name=username&amp;rabbitmq.server.password=password&amp;rabbitmq.queue.name=RabbitMQTestQueue&amp;rabbitmq.queue.routing.key=RabbitMQTestQueue&amp;rabbitmq.server.virtual.host=hostname&amp;rabbitmq.exchange.name=amq.direct">
        <suspendOnFailure>
            <initialDuration>-1</initialDuration>
            <progressionFactor>-1</progressionFactor>
        </suspendOnFailure>
        <markForSuspension>
            <retriesBeforeSuspension>0</retriesBeforeSuspension>
        </markForSuspension>
    </address>
</endpoint>

API Hit:

postman

RabbitMQ Console:

rabbitmq

user_99071820
Related