Complete End-to-End Integration Example

Overview

A complete working example showing every file needed for a Customer Extension (CX) import integration: route, mapper, filter, configuration properties, and integration test. Use this as a reference template when building new integrations.

Scenario

Import customer segment data from a CSV file into a Pricefx Customer Extension table called "Segments". After import, verify the data by exporting with a filter.

Files

routes/import-customer-segments.xml

XML

<routes xmlns="http://camel.apache.org/schema/spring">
    <route id="import-customer-segments">
        <from uri="file:{{import.cx.directory}}?{{archive.file}}&amp;{{read.lock}}"/>
        <log message="Starting CX Segments import: ${header.CamelFileNameOnly}" loggingLevel="INFO"/>
        <to uri="pfx-csv:streamingUnmarshal?skipHeaderRecord=true&amp;useReusableParser=true"/>
        <to uri="pfx-api:loaddataFile?objectType=CX&amp;mapper=import-customer-segments.mapper&amp;batchSize=200000&amp;businessKeys=customerId"/>
        <log message="CX Segments import completed: ${header.CamelFileNameOnly}" loggingLevel="INFO"/>
        <onCompletion onCompleteOnly="true">
            <log message="Import successful, verifying record count" loggingLevel="INFO"/>
            <to uri="pfx-api:fetch?objectType=CX&amp;filter=cx-segments.filter&amp;countOnly=true"/>
            <log message="CX Segments total records: ${body}" loggingLevel="INFO"/>
        </onCompletion>
        <onCompletion onFailureOnly="true">
            <log message="Import FAILED for file: ${header.CamelFileNameOnly}" loggingLevel="ERROR"/>
        </onCompletion>
    </route>
</routes>

mappers/import-customer-segments.mapper.xml

XML

<mappers>
    <loadMapper id="import-customer-segments.mapper">
        <constant expression="Segments" out="name"/>
        <body in="customerId" out="customerId"/>
        <body in="segment" out="attribute1"/>
        <body in="tier" out="attribute2"/>
        <body in="annualRevenue" out="attribute3" converterExpression="stringToDecimal"/>
        <body in="contractStartDate" out="attribute4" converterExpression="stringToDate"/>
        <body in="accountManager" out="attribute5"/>
        <body in="region" out="attribute6"/>
        <body in="isActive" out="attribute7" converterExpression="stringToBoolean"/>
    </loadMapper>
</mappers>

filters/cx-segments.filter.xml

XML

<filters>
    <filter id="cx-segments.filter">
        <and>
            <criterion fieldName="name" operator="equals" value="Segments"/>
        </and>
        <resultFields>customerId,attribute1,attribute2,attribute3</resultFields>
        <sortBy>customerId</sortBy>
    </filter>
</filters>

config/application.properties (additions)

# Customer Extension import
import.cx.directory=/data/imports/customer-extensions

# File consumer properties (define once, reuse everywhere)
archive.file=move=.archive/%24%7Bdate:now:yyyy%7D/%24%7Bdate:now:MM%7D/%24%7Bfile:name.noext%7D__%24%7Bdate:now:yyyyMMdd_HHmmss%7D.%24%7Bfile:ext%7D
read.lock=readLock=changed

Test CSV: test-data/customer-segments.csv

customerId,segment,tier,annualRevenue,contractStartDate,accountManager,region,isActive
CUST-001,Enterprise,Gold,1500000.00,2024-01-15,John Smith,EMEA,true
CUST-002,SMB,Silver,250000.00,2024-03-01,Jane Doe,APAC,true
CUST-003,Enterprise,Platinum,5000000.00,2023-06-20,Bob Wilson,NA,true
CUST-004,SMB,Bronze,50000.00,2024-07-10,Alice Brown,EMEA,false

Test: src/test/groovy/com/example/ImportCustomerSegmentsSpec.groovy

Groovy

import com.pricefx.integrationmanager.testing.IntegrationTestSpecification

class ImportCustomerSegmentsSpec extends IntegrationTestSpecification {

    def "should import customer segments from CSV into CX table"() {
        given: "a CSV file with customer segment data"
        def csvContent = """\
customerId,segment,tier,annualRevenue,contractStartDate,accountManager,region,isActive
CUST-001,Enterprise,Gold,1500000.00,2024-01-15,John Smith,EMEA,true
CUST-002,SMB,Silver,250000.00,2024-03-01,Jane Doe,APAC,true"""
        writeToImportDirectory("customer-segments.csv", csvContent)

        and: "Pricefx loaddataFile endpoint is mocked"
        mockPricefxLoadDataFile(objectType: "CX") {
            respondWith(status: 200, body: '{"response":{"data":[]}}')
        }

        when: "the import route runs"
        triggerRoute("import-customer-segments")
        waitForRouteCompletion("import-customer-segments")

        then: "loaddataFile was called with CX object type"
        verifyPricefxLoadDataFileCalled(objectType: "CX", times: 1)
    }

    def "should map fields correctly including type conversions"() {
        given: "a CSV file with typed fields"
        def csvContent = """\
customerId,segment,tier,annualRevenue,contractStartDate,accountManager,region,isActive
CUST-010,Enterprise,Gold,1500000.00,2024-01-15,Test Manager,EMEA,true"""
        writeToImportDirectory("customer-segments.csv", csvContent)

        and: "Pricefx endpoint is mocked and captures payload"
        mockPricefxLoadDataFile(objectType: "CX") {
            respondWith(status: 200, body: '{"response":{"data":[]}}')
        }

        when: "the import route runs"
        triggerRoute("import-customer-segments")
        waitForRouteCompletion("import-customer-segments")

        then: "the mapper set the CX table name and converted types"
        def payload = getCapturedLoadDataFilePayload(objectType: "CX")
        payload[0].name == "Segments"
        payload[0].customerId == "CUST-010"
        payload[0].attribute1 == "Enterprise"
        payload[0].attribute3 instanceof BigDecimal
    }
}

File Naming Convention

File	Name Pattern	ID Pattern
Route	`routes/{route-name}.xml`	`id="{route-name}"`
Mapper	`mappers/{route-name}.mapper.xml`	`id="{route-name}.mapper"`
Filter	`filters/{filter-name}.filter.xml`	`id="{filter-name}.filter"`
Properties	`config/application.properties`	N/A
Test	`src/test/groovy/.../Spec.groovy`	N/A

Route ID must match filename -- import-customer-segments.xml -> id="import-customer-segments". Mismatch causes deployment failure.

Mapper ID must match filename -- import-customer-segments.mapper.xml -> id="import-customer-segments.mapper".

How It Works

File Pickup: file: monitors {{import.cx.directory}} for CSV files. {{archive.file}} moves processed files to timestamped archive. {{read.lock}} waits until file is fully written.
Streaming Parse: pfx-csv:streamingUnmarshal parses CSV without loading full file into memory. Must be BEFORE any split.
Upload to Pricefx: pfx-api:loaddataFile streams parsed data to the server. objectType=CX targets Customer Extensions. businessKeys=customerId enables upsert by customer ID.
Mapper: <constant expression="Segments" out="name"/> sets the CX table name -- mandatory for PX/CX. Type converters handle annualRevenue (decimal), contractStartDate (date), and isActive (boolean).
Filter: Used in onCompletion to verify record count after import. The resultFields and sortBy optimize the verification query.
On Success: Logs record count via a filtered fetch with countOnly=true.
On Failure: Logs error with the filename that failed.

Common Pitfalls

Missing <constant out="name"/>: CX imports MUST include the table name constant in the mapper. Without it, the import fails with IllegalStateException.
Mapper filename convention: Always use .mapper.xml suffix. The ID must match the full filename minus .xml.
streamingUnmarshal inside split: Will fail. It must be BEFORE any <split> element.
Missing businessKeys for CX: Without businessKeys=customerId, every import creates new records instead of updating.
Filter name field: For CX/PX, the name field in the filter refers to the extension table name, not a customer name.