How to enrich payload and insert objects into MongoDB using Mule

Question

I have a message enricher in my Mule flow that looks like this:

<set-variable doc:name="Variable" value="#[payload['MeterUID']]" variableName="#['theKey']"/>
    <enricher target="#[payload]" doc:name="Message Enricher">
<mongo:find-objects-using-query-map config-ref="Mongo_DB1" collection="meterentity" doc:name="Mongo DB">
    <mongo:query-attributes>
                <mongo:query-attribute key="_id">#[theKey]</mongo:query-attribute>
    </mongo:query-attributes>
    <mongo:fields>
        <mongo:field>IpAddress</mongo:field>
        <mongo:field>LastSetTime</mongo:field>
        <mongo:field>LastReadGsmData</mongo:field>
    </mongo:fields>
</mongo:find-objects-using-query-map>

and this is the payload that the message enricher recieves from my inbound-endpoint:

{TimeStamp=2013-12-16 08:48:33,270, MeterUID=4B414D000000011613CF, SignalStrengthIndication=15, CellID=4938, LocationAreaCode=280, MobileCountryCode=238, MobileNetworkCode=1}

at the Mongo-collection to json my payload looks like this:

[ { "_id" : "4B414D000000011613CC" , "IpAddress" : "10.12.189.12" , "LastSetTime" : { "$date" : "2014-03-11T14:40:36.987Z"} , "LastReadGsmData" : { "$date" : "2014-03-11T14:40:47.253Z"}}]

This seems wrong to me. First, where did my my first payload go ? And second I get this this error when the flow tryes to inserted into my mongodb:

Exception stack is:
1. BasicBSONList can only work with numeric keys, not: [_id] (java.lang.IllegalArgumentException)
  org.bson.types.BasicBSONList:161 (null)
2. Failed to invoke insertObject. Message payload is of type: String

How can I make this work? I am still vary new to mule so i hope you guys can help me here.

My entire flow:

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

<mule xmlns:vm="http://www.mulesoft.org/schema/mule/vm" xmlns:json="http://www.mulesoft.org/schema/mule/json" xmlns:mulexml="http://www.mulesoft.org/schema/mule/xml" xmlns:mongo="http://www.mulesoft.org/schema/mule/mongo" xmlns:tracking="http://www.mulesoft.org/schema/mule/ee/tracking" xmlns:data-mapper="http://www.mulesoft.org/schema/mule/ee/data-mapper" xmlns:file="http://www.mulesoft.org/schema/mule/file" 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" version="EE-3.4.1"
    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/file http://www.mulesoft.org/schema/mule/file/current/mule-file.xsd
http://www.mulesoft.org/schema/mule/ee/data-mapper http://www.mulesoft.org/schema/mule/ee/data-mapper/current/mule-data-mapper.xsd
http://www.mulesoft.org/schema/mule/ee/tracking http://www.mulesoft.org/schema/mule/ee/tracking/current/mule-tracking-ee.xsd
http://www.mulesoft.org/schema/mule/mongo http://www.mulesoft.org/schema/mule/mongo/2.0/mule-mongo.xsd
http://www.mulesoft.org/schema/mule/xml http://www.mulesoft.org/schema/mule/xml/current/mule-xml.xsd
http://www.mulesoft.org/schema/mule/json http://www.mulesoft.org/schema/mule/json/current/mule-json.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/current/mule-vm.xsd">
    <data-mapper:config name="csv_to_xml_7" transformationGraphPath="csv_to_xml_7.grf" doc:name="csv_to_xml_7"/>
    <data-mapper:config name="csv_to_xml_8" transformationGraphPath="csv_to_xml_8.grf" doc:name="csv_to_xml_8"/>
    <mongo:config name="Mongo_DB" username="$[admin]" doc:name="Mongo DB">
        <mongo:connection-pooling-profile initialisationPolicy="INITIALISE_ONE" exhaustedAction="WHEN_EXHAUSTED_GROW"/>
    </mongo:config>
    <mongo:config name="Mongo_DB1" username="$[admin]" doc:name="Mongo DB">
        <mongo:connection-pooling-profile initialisationPolicy="INITIALISE_ONE" exhaustedAction="WHEN_EXHAUSTED_GROW"/>
    </mongo:config>
    <data-mapper:config name="xml_to_json_2" transformationGraphPath="xml_to_json_2.grf" doc:name="xml_to_json_2"/>
    <data-mapper:config name="xml_to_json_3" transformationGraphPath="xml_to_json_3.grf" doc:name="xml_to_json_3"/>
    <mongo:config name="Mongo_DB2" username="$[admin]" doc:name="Mongo DB">
        <mongo:connection-pooling-profile initialisationPolicy="INITIALISE_ONE" exhaustedAction="WHEN_EXHAUSTED_GROW"/>
    </mongo:config>
    <flow name="p2pcontrollerloganalyserFlow1" doc:name="p2pcontrollerloganalyserFlow1">
        <file:inbound-endpoint path="C:\Users\Simon\Desktop\CSVFile" responseTimeout="10000" doc:name="File" moveToDirectory="C:\Users\Simon\Desktop\CSVPros" />
        <byte-array-to-string-transformer doc:name="Byte Array to String"/>
        <choice doc:name="Choice">
            <when expression="#[message.outboundProperties.originalFilename=='gsmdata.log']">
                <data-mapper:transform config-ref="csv_to_xml_7" doc:name="GSMV1 to xml"/>
                <splitter expression="#[xpath('//Root')]" doc:name="Splitter"/>
                <mulexml:dom-to-xml-transformer doc:name="DOM to XML"/>
                <data-mapper:transform config-ref="xml_to_json_2" doc:name="XML To JSON"/>
                <byte-array-to-string-transformer doc:name="Byte Array to String"/>
                <vm:outbound-endpoint exchange-pattern="one-way" path="json1" doc:name="VM"/>
            </when>
            <when expression="#[message.outboundProperties.originalFilename=='gsmdatav2.log']">
                <data-mapper:transform config-ref="csv_to_xml_8" doc:name="GSMV2 to XML"/>
                <splitter expression="#[xpath('//Root')]" doc:name="Splitter"/>
                <mulexml:dom-to-xml-transformer doc:name="DOM to XML"/>
                <data-mapper:transform config-ref="xml_to_json_3" doc:name="XML To JSON"/>
                <byte-array-to-string-transformer doc:name="Byte Array to String"/>
                <vm:outbound-endpoint exchange-pattern="one-way" path="json2" doc:name="VM"/>
            </when>
            <otherwise>
                <file:outbound-endpoint path="C:\Users\Simon\Desktop\CSVFile" responseTimeout="10000" doc:name="Failed GSM"/>
            </otherwise>
        </choice>
    </flow>
    <flow name="json1persistent" doc:name="json1persistent">
        <vm:inbound-endpoint exchange-pattern="one-way" path="json1" doc:name="VM"/>
        <json:json-to-object-transformer returnClass="java.lang.Object" doc:name="JSON to Object"/>
        <set-variable doc:name="Variable" value="#[payload['MeterUID']]" variableName="#['theKey']"/>
        <enricher target="#[payload]" doc:name="Message Enricher">
    <mongo:find-objects-using-query-map config-ref="Mongo_DB1" collection="meterentity" doc:name="Mongo DB">
        <mongo:query-attributes>
                    <mongo:query-attribute key="_id">#[theKey]</mongo:query-attribute>
        </mongo:query-attributes>
        <mongo:fields>
            <mongo:field>IpAddress</mongo:field>
            <mongo:field>LastSetTime</mongo:field>
            <mongo:field>LastReadGsmData</mongo:field>
        </mongo:fields>
    </mongo:find-objects-using-query-map>
</enricher>
        <mongo:mongo-collection-to-json doc:name="Mongo DB"/>
        <mongo:insert-object config-ref="Mongo_DB2" collection="GSMdata" doc:name="Mongo DB"/>
        <!-- <foreach collection="#[payload]" doc:name="For Each">
            <mongo:insert-object config-ref="Mongo_DB" collection="GSMdata" doc:name="Mongo DB"/>
        </foreach> --> 
        <!--  <enricher doc:name="Message Enricher">
            <mongo:get-file-content config-ref="Mongo_DB" query-ref="#[payload['MeterUID']]" doc:name="Find Meter"/>
        </enricher>-->
    </flow>
</mule>

I am tying to read a CSV file, splitting it into smaller messages, enrich each of thise messages with the data from a MongoDB and finally insert each message into another MongoDB.

Anton Kupias Anton Kupias · Accepted Answer · 2014-05-26T19:44:09

First, where did my my first payload go ?

enricher target="#[payload]" means that you are setting your current payload to whatever is returned by the enricher. You really want the target to be something else, such as a variable: target="#[variable:myVar]".

second I get this this error when the flow tryes to inserted into my mongodb

You are not sharing any of your relevant configuration here, but I guess you are now sending your list of queried Mongo objects to some component that expects a single Map object. Please refer to your first problem.

If your aim is to add the fields from queried Mongo objects to your current payload, you should use a variable as the enrich target like in my above example. You can then later combine the payload and the enriched variable containing the Mongo object(s).

Further, if you are only getting a single object from Mongo, you can extract it from the Mongo list by using something like source="#[payload.toArray()[0]]" in the enricher component.

As your payload and the myVar variable are now both Java Maps, you can combine them with plain Java:

<expression-component doc:name="Expression">payload.putAll(myVar)</expression-component>

If you want to get rid of the _id field returned by Mongo, you can do it with

<expression-component doc:name="Expression">payload.remove('_id')</expression-component>

EDIT: Now that your payload is a Map, you can insert it into Mongo with

<mongo:insert-object-from-map config-ref="Mongo_DB2" collection="GSMdata">
   <mongo:element-attributes ref="#[payload]"/> 
</mongo:insert-object-from-map>

How to enrich payload and insert objects into MongoDB using Mule

1 Answers