Add a Mail Ingress Connector with the configurations shown below (assuming a Gmail email address; replace account details as appropriate)

Next add a Logger to generate a log before attempting to save the attachment.

As the file content does not contain any reference to the originating factory, we shall save it by the name of the factory itself (i.e. the value of X-Zenythz-Factory header), for correlation during further processing. For this, add an Attachment Saver with the configuration shown below.

Finally, complete the flow with a Successful Flow End element.

Add a NIO File Ingress Connector to pick the saved report.

After the file is picked, its name (which is now the factory ID) will be available via the message header ultra.file.name. Add a Logger to log this ID before we start processing the report.

As the machine entries in the report can be processed as individual database update queries, we can use an XML splitter to process them parallely. As we need an XML payload for this, let’s first add a CSV-to-XML transformer to convert the CSV content to XML.

<factory>
    <machine>
        <id>1</id>
        <hours>10</hours>
        ...
    </machine>
    <machine>
    ...
</factory>

Next add an XML Splitter to split the converted payload at nodes denoted by the XPath /factory/machine. From this point onwards, until an XML Aggregator is encountered on the flow, the processing element operations would be applied identically on the split payload segments being propagated in individual messages.

For logging purposes, let’s first extract the machine ID into a variable using an XPath String Extractor.

Next add a Logger to log the machine ID, as a confirmation before performing the actual database update.

For database access, a JDBC data source has to be added as a project resource. Switch to the project.xpml and right click on the XML content and from the context menu select Resource Template. Next from the pop-up window select Daatasource and click next. After that fill the necessary values as shown in below table and click save button. We shall assume a MySQL database configured on port 3306 on localhost.

We can simplify the database update by utilizing a Database Mapping File to directly map the required data from the message into placeholders in a predefined parameterized query, rather than constructing the query by hand. For this, add a mapping file report-mapper.dson by right-clicking the resources directory on the Project Structure pane and selecting New → DB Configuration, and populate it with the following JSON content:

{
  "messageType" : "XML",
  "columnNamesPresent" : false,
  "parameters" : [ {
    "type" : "PATH",
    "name" : "hours",
    "path" : "/machine/hours"
  }, {
    "type" : "PATH",
    "name" : "restarts",
    "path" : "/machine/restarts"
  }, {
    "type" : "PATH",
    "name" : "minor_repairs",
    "path" : "/machine/minor_repairs"
  }, {
    "type" : "PATH",
    "name" : "major_repairs",
    "path" : "/machine/major_repairs"
  }, {
    "type" : "PATH",
    "name" : "extra_cost",
    "path" : "/machine/extra_cost"
  }, {
    "type" : "VARIABLE",
    "name" : "machine_id",
    "variableName" : "machine_id",
    "className" : "java.lang.String",
    "propagateToParent" : false
  }, {
    "type" : "HEADER",
    "name" : "factory_id",
    "headerName" : "ultra.file.name",
    "className" : "java.lang.String"
  } ]
}

Now add a Database Processor element to perform the actual database update query, using the template we just defined.

Before we finish the flow, we need to aggregate the split message paths using an XML Aggregator (although the aggregation result would not be useful for us in this case).

Now end the flow using a Successful Flow End element.