MuleSoft Mastery: From Zero to Hero | Ch.10

<jms:config name="JMS_Config">
  <jms:active-mq-connection>
    <jms:factory-configuration brokerUrl="tcp://broker:61616"/>
  </jms:active-mq-connection>
</jms:config>

<flow name="publish-order">
  <http:listener config-ref="HTTP" path="/orders"/>
  <!-- destinationType defaults to QUEUE; set TOPIC to broadcast -->
  <jms:publish config-ref="JMS_Config" destination="orders.in"/>
</flow>

<flow name="consume-order">
  <jms:listener config-ref="JMS_Config" destination="orders.in"
                ackMode="MANUAL" transactionalAction="ALWAYS_BEGIN">
  </jms:listener>

  <flow-ref name="persist-order"/>

  <!-- only ack once processing succeeded; errors roll back & redeliver -->
  <jms:ack ackId="#[attributes.ackId]"/>
</flow>

<flow name="process-inbound-files">
  <file:listener config-ref="File_Config" directory="/data/in"
                 moveToDirectory="/data/archive" autoDelete="false"
                 timeBetweenSizeCheck="2" timeBetweenSizeCheckUnit="SECONDS">
    <scheduling-strategy><fixed-frequency frequency="10000"/></scheduling-strategy>
    <file:matcher filenamePattern="*.csv"/>   <!-- only .csv -->
  </file:listener>

  <flow-ref name="parse-and-load"/>

  <error-handler>
    <on-error-continue type="ANY">
      <!-- on failure the auto-archive is skipped, so move it ourselves -->
      <file:move config-ref="File_Config"
                 sourcePath="#[attributes.path]"
                 targetPath="/data/error"/>
    </on-error-continue>
  </error-handler>
</flow>

<sftp:config name="SFTP_Config">
  <sftp:connection host="sftp.partner.com" port="22"
                   username="acme" password="${sftp.password}">
    <!-- reuse connections instead of opening one per operation -->
    <pooling-profile maxActive="5" maxIdle="2"
                     exhaustedAction="WHEN_EXHAUSTED_WAIT"/>
    <!-- retry transient network failures -->
    <reconnection>
      <reconnect frequency="5000" count="3"/>
    </reconnection>
  </sftp:connection>
</sftp:config>

<flow name="sftp-poll">
  <sftp:listener config-ref="SFTP_Config" directory="/orders/in"
                 autoDelete="false" timeBetweenSizeCheck="2">
    <scheduling-strategy><fixed-frequency frequency="15000"/></scheduling-strategy>
    <sftp:matcher filenamePattern="order-*.csv"/>
  </sftp:listener>
  <flow-ref name="parse-and-load"/>
</flow>

<!-- SOQL query → returns a list of records -->
<salesforce:query config-ref="Salesforce_Config">
  <salesforce:salesforce-query>
    SELECT Id, Name, Email FROM Contact WHERE Account.Name = ':acct'
  </salesforce:salesforce-query>
  <salesforce:parameters>#[{ 'acct': payload.account }]</salesforce:parameters>
</salesforce:query>

<!-- create expects an ARRAY of records, even for a single one -->
<salesforce:create type="Contact" config-ref="Salesforce_Config">
  <salesforce:records>#[[
    {
      LastName: payload.lastName,
      Email: payload.email,
      AccountId: payload.accountId
    }
  ]]</salesforce:records>
</salesforce:create>

<salesforce:upsert type="Account"
                   externalIdFieldName="External_Id__c"
                   config-ref="Salesforce_Config">
  <salesforce:records>#[[
    {
      External_Id__c: payload.externalId,
      Name: payload.name,
      Type: payload.type
    }
  ]]</salesforce:records>
</salesforce:upsert>

<flow name="on-new-case">
  <salesforce:subscribe-channel-listener config-ref="Salesforce_Config"
        streamingChannel="/topic/NewCases"/>
  <logger level="INFO" message="#[payload.payload.CaseNumber]"/>
</flow>

<flow name="process-once">
  <jms:listener config-ref="JMS_Config" destination="orders.in" ackMode="MANUAL"/>

  <os:contains objectStore="processedIds" key="#[attributes.properties.JMSMessageID]"
               target="seen"/>

  <choice>
    <when expression="#[not vars.seen]">
      <flow-ref name="do-the-work"/>
      <os:store objectStore="processedIds"
                key="#[attributes.properties.JMSMessageID]">
        <os:value>#[now()]</os:value>
      </os:store>
    </when>
    <otherwise>
      <logger level="INFO" message="Duplicate — skipping"/>
    </otherwise>
  </choice>

  <jms:ack ackId="#[attributes.ackId]"/>
</flow>

<flow name="sftp-order-ingest">
  <sftp:listener config-ref="SFTP_Config" directory="/orders/in"
                 moveToDirectory="/orders/archive" timeBetweenSizeCheck="2">
    <scheduling-strategy><fixed-frequency frequency="15000"/></scheduling-strategy>
    <sftp:matcher filenamePattern="order-*.csv"/>
  </sftp:listener>

  <!-- CSV → array of order objects -->
  <ee:transform>
    <ee:message><ee:set-payload><![CDATA[%dw 2.0
output application/json
---
payload map (row) -> {
  orderId:       row.orderId,
  customerEmail: row.email,
  total:         row.total as Number,
  sku:           row.sku
}]]></ee:set-payload></ee:message>
  </ee:transform>

  <foreach collection="#[payload]">
    <flow-ref name="validate-order"/>
    <jms:publish config-ref="JMS_Config" destination="orders.in"/>
  </foreach>

  <error-handler>
    <on-error-continue type="ANY">
      <sftp:move config-ref="SFTP_Config"
                 sourcePath="#['/orders/in/' ++ attributes.name]"
                 targetPath="#['/orders/error/' ++ attributes.name]"/>
    </on-error-continue>
  </error-handler>
</flow>

<flow name="jms-order-consumer">
  <jms:listener config-ref="JMS_Config" destination="orders.in"
                ackMode="MANUAL" transactionalAction="ALWAYS_BEGIN"/>

  <flow-ref name="enrich-with-product"/>

  <!-- upsert is idempotent: replaying converges to the same record -->
  <salesforce:upsert type="Order__c" externalIdFieldName="Order_ID__c"
                     config-ref="Salesforce_Config">
    <salesforce:records>#[[
      {
        Order_ID__c:       payload.orderId,
        Customer_Email__c: payload.customerEmail,
        Total_Amount__c:   payload.total,
        Status__c:         "New"
      }
    ]]</salesforce:records>
  </salesforce:upsert>

  <jms:ack ackId="#[attributes.ackId]"/>

  <error-handler>
    <!-- rollback → broker redelivers; after maxRedelivery it routes to the DLQ -->
    <on-error-propagate type="ANY"/>
  </error-handler>
</flow>