---
language: "en"
---
# Routes

*** ** * ** ***

## Routes

Camel route definitions --- the main integration logic. Each file defines one or more routes that orchestrate data flow.  

| **Attribute** |                                     **Details**                                      |
|---------------|--------------------------------------------------------------------------------------|
| **Purpose**   | Define data flow: what triggers the route, how data is processed, where it goes      |
| **Format**    | XML --- \<routes\> wrapper with Camel Spring DSL \<route\> elements inside           |
| **Naming**    | Descriptive, by function: import-products.xml, export-prices.xml, event-handlers.xml |
| **Loaded by** | Camel context at startup --- all \*.xml files are auto-discovered                    |

*** ** * ** ***

## When to Use Each Pattern

|           Pattern            |             Use when             |                          Key components                          |
|------------------------------|----------------------------------|------------------------------------------------------------------|
| CSV → Pricefx (loaddataFile) | Default for P, PX, CX, C imports | `streamingUnmarshal` + `loaddataFile`                            |
| CSV → Pricefx (loaddata)     | Need Groovy row-level logic      | `split+tokenize` + `unmarshal` + `loaddata`                      |
| CSV → Data Source (DMDS)     | PA/Datamart imports              | `split+tokenize` + `unmarshal` + `loaddata` + `flush`            |
| Pricefx → CSV                | Export data                      | `fetch` + `marshal` + `file:`                                    |
| Pricefx → CSV (large)        | Export 50k+ rows                 | `fetch?batchedMode=true` + `split` + `fetchIterator` + `marshal` |
| Event handler                | React to Pricefx events          | `direct:` from event mapping in properties                       |
| Scheduled                    | Periodic execution               | `quartz://` with cron expression                                 |

*** ** * ** ***

## Route XML Structure

Routes are defined in XML files under the routes/ directory using Apache Camel Spring DSL with Pricefx extensions. Each route file uses a \<routes\> wrapper element with the Camel Spring namespace --- no beans wrapper or XML declaration is needed.

### Minimal Route File

XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="myRoute">
            <from uri="..."/>
            <to uri="..."/>
        </route>
    </routes>

*** ** * ** ***

## Common Route Patterns

### Import: CSV File to Pricefx (Data Source / DMDS)

The standard DMDS import pattern: read CSV, split into batches, unmarshal, load, flush.
XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="csvImportToDatasource">
            <from uri="file:{{import.fromUri}}"/>
            <split>
                <tokenize group="5000" token="&#10;"/>
                <to uri="pfx-csv:unmarshal?header=sku,label,price&amp;skipHeaderRecord=true&amp;delimiter=,"/>
                <to uri="pfx-api:loaddata?mapper=myMapper&amp;objectType=DM&amp;dsUniqueName=Product"/>
            </split>
            <onCompletion onCompleteOnly="true">
                <to uri="pfx-api:flush?dataFeedName=DMF.Product&amp;dataSourceName=DMDS.Product"/>
            </onCompletion>
        </route>
    </routes>

**Key points:**

* split + tokenize group="5000" batches CSV rows for efficient loading

* pfx-csv:unmarshal converts CSV text to List of Maps

* pfx-api:loaddata sends data to Pricefx using a mapper

* onCompletion with pfx-api:flush is mandatory for DMDS imports

* Both `objectType=DM` and `objectType=DMDS` work for loaddata with `dsUniqueName`, but `DM` is the standard convention used across the project

### Export: Pricefx to CSV File

XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="exportToCsv">
            <from uri="timer://fetchData?repeatCount=1"/>
            <to uri="pfx-api:fetch?filter=myFilter&amp;objectType=PX"/>
            <to uri="pfx-csv:marshal"/>
            <to uri="file:export?fileName=data.csv&amp;fileExist=Append"/>
        </route>
    </routes>

### Batched Export (Large Datasets)

For large datasets, use batchedMode to avoid memory issues:
XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="batchedExport">
            <from uri="timer://fetchData?repeatCount=1"/>
            <to uri="pfx-api:fetch?filter=myFilter&amp;objectType=PX&amp;batchedMode=true&amp;batchSize=5000"/>
            <split>
                <simple>${body}</simple>
                <to uri="pfx-api:fetchIterator"/>
                <to uri="pfx-csv:marshal"/>
                <to uri="file:export?fileName=data.csv&amp;fileExist=Append"/>
            </split>
        </route>
    </routes>

The first pfx-api:fetch (with batchedMode=true) returns batch descriptors. The split iterates them, and pfx-api:fetchIterator fetches each batch using state from the exchange.

### Import: Product Master Data (P)

XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="productImport">
            <from uri="file:{{products.fromUri}}"/>
            <to uri="pfx-csv:unmarshal"/>
            <to uri="pfx-api:loaddata?objectType=P&amp;mapper=productMapper"/>
            <onCompletion onCompleteOnly="true">
                <to uri="pfx-api:internalCopy?label=Product"/>
            </onCompletion>
        </route>
    </routes>

### Delta / Incremental Export

Export only records modified since last run. The filter is defined in its own file under filters/:

**filters/delta-filter.xml:**
XML

    <filters>
        <filter id="deltaFilter">
            <and>
                <criterion fieldName="name" operator="equals" value="Cars"/>
                <criterion fieldName="lastUpdateDate" operator="greaterThan" value="simple:${header.lastExportTimestamp}"/>
            </and>
        </filter>
    </filters>

**routes/delta-export.xml:**
XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="deltaExport">
            <from uri="timer://deltaExport?repeatCount=1"/>
            <to uri="pfx-api:fetch?filter=deltaFilter&amp;objectType=PX&amp;batchedMode=true&amp;batchSize=5000"/>
            <split>
                <simple>${body}</simple>
                <to uri="pfx-api:fetchIterator"/>
                <to uri="pfx-csv:marshal"/>
                <to uri="file:export?fileName=delta-${date:now:yyyyMMdd-HHmmss}.csv"/>
            </split>
        </route>
    </routes>

### Export to SFTP

XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="exportToSftp">
            <from uri="timer://fetchData?repeatCount=1"/>
            <to uri="pfx-api:fetch?filter=myFilter&amp;objectType=PX"/>
            <to uri="pfx-csv:marshal"/>
            <to uri="sftp:{{ftp.address}}:{{ftp.port}}?username={{ftp.username}}&amp;password={{ftp.password}}&amp;fileName={{ftp.path}}/export_${date:now:yyyyMMdd}.csv&amp;useUserKnownHostsFile=false"/>
        </route>
    </routes>

*** ** * ** ***

## Events

Pricefx generates events that IM can process. Configure pull-based event handling in application.properties:

### Pull Events (Recommended)

    integration.events.enabled=true
    integration.events.scheduler-delay=60000
    integration.events.delay=10000
    integration.events.event-to-route-mapping.PADATALOAD_COMPLETED=direct:eventPADataLoadCompleted
    integration.events.event-to-route-mapping.CALCULATION_COMPLETED_CFS=direct:eventCFSCompleted

Event handler route:
XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="eventPADataLoadCompleted">
            <from uri="direct:eventPADataLoadCompleted"/>
            <setHeader headerName="targetName">
                <simple>${body[data][0][targetName]}</simple>
            </setHeader>
            <choice>
                <when>
                    <simple>${headers.targetName} == "DMDS.Product"</simple>
                    <to uri="pfx-api:truncate?targetName=DMF.Product"/>
                </when>
            </choice>
        </route>
    </routes>

### Event Properties

|             **Property**             | **Default** |           **Description**            |
|--------------------------------------|-------------|--------------------------------------|
| `integration.events.enabled`         | false       | Enable/disable event processing      |
| `integration.events.scheduler-delay` | 60000       | Pull interval from Pricefx (ms)      |
| `integration.events.delay`           | 10000       | File read interval (ms)              |
| `integration.events.max-events`      | 1000        | Max events per pull                  |
| `integration.events.connection`      | pricefx     | Pricefx connection name              |
| `integration.events.initial-delay`   | 1000        | Initial delay before first pull (ms) |

### Stuck Events Monitoring

|                         **Property**                         | **Default** |        **Description**         |
|--------------------------------------------------------------|-------------|--------------------------------|
| `integration.events.check-stuck-processing.enabled`          | true        | Check for stuck events         |
| `integration.events.check-stuck-processing.scheduled-ms`     | 7200000     | Check interval (2 hours)       |
| `integration.events.check-stuck-processing.timeout-duration` | 24          | When event is considered stuck |
| `integration.events.check-stuck-processing.timeout-unit`     | h           | Unit: d, h, m, s               |

*** ** * ** ***

## File Consumer Patterns

When reading files from the local file system, always configure archiving and file locking via standard properties.

### Standard Properties

These properties are defined once in `config/application.properties` and referenced in route URIs via `{{...}}` placeholders:

    # Move processed files to timestamped archive (default --- always add this)
    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

    # Wait until file size stabilizes before processing (use when no .done marker)
    read.lock=readLock=changed

    # Wait for a .done marker file before processing (use when upstream writes .done)
    done.file=doneFileName=%24%7Bfile:name%7D.done

    # Move files that fail processing to an error folder (optional, add on request)
    error.file=moveFailed=.error/%24%7Bfile:name.noext%7D__%24%7Bdate:now:yyyyMMdd-HHmmss%7D.%24%7Bfile:ext%7D

### File URI Template

XML

    <from uri="file://{{integration.sftp.root}}/my-path?delay=10000&amp;{{archive.file}}&amp;{{read.lock}}"/>

**Rules:**

* Always include `{{archive.file}}` --- moves processed files to `.archive/YYYY/MM/filename__timestamp.ext`

* Use **either** `{{read.lock}}` **or** `{{done.file}}` --- never both, never neither. Default: `{{read.lock}}`

  * `{{read.lock}}` --- waits until the file size stabilizes (no external signal required)

  * `{{done.file}}` --- waits for a `.done` marker file (e.g., `data.csv.done`) before reading `data.csv`

* `{{error.file}}` is optional --- moves failed files to `.error/filename__timestamp.ext` on processing failure

* Never use `noop=true` on `file:` consumers --- files must be moved/archived after processing (exception: `pollEnrich` for one-shot reads)

* Never use `include` parameter by default

*** ** * ** ***

## Scheduled Exports with Quartz Cron

For time-based scheduling, use the Quartz scheduler. In Camel URIs, **spaces in cron expressions are replaced with** `+`:
XML

    <!-- Every day at 6:00 AM -->
    <from uri="quartz://export/my-export?cron=0+0+6+*+*+?"/>

    <!-- Every day at midnight -->
    <from uri="quartz://export/my-export?cron=0+0+0+*+*+?"/>

    <!-- Every hour -->
    <from uri="quartz://export/my-export?cron=0+0+*+*+*+?"/>

    <!-- Monday--Friday at 8:00 AM -->
    <from uri="quartz://export/my-export?cron=0+0+8+?+*+MON-FRI"/>

    <!-- First day of each month at 1:00 AM -->
    <from uri="quartz://export/my-export?cron=0+0+1+1+*+?"/>

Cron format: `seconds minutes hours day-of-month month day-of-week`

*** ** * ** ***

## Delta Sync Pattern

Delta sync exports only records modified since the last run, using `pfx-config:get/set` to persist the timestamp between runs.
XML

    <routes xmlns="http://camel.apache.org/schema/spring">
        <route id="delta-export-products">
            <from uri="quartz://export/delta-export-products?cron=0+0+6+*+*+?"/>

            <!-- Read last export timestamp from partition -->
            <toD uri="pfx-config:get?name={{integration.name}}.${routeId}.export.timestamp&amp;toHeader=lastExportTimestamp"/>

            <!-- Fallback: first run exports everything -->
            <choice>
                <when>
                    <simple>${headers.lastExportTimestamp} == null || ${headers.lastExportTimestamp} == ''</simple>
                    <setHeader name="lastExportTimestamp">
                        <constant>1970-01-01T00:00:00</constant>
                    </setHeader>
                </when>
            </choice>

            <!-- Capture current time as upper bound -->
            <setHeader name="currentExportTimestamp">
                <simple>${date-with-timezone:now:UTC:yyyy-MM-dd'T'HH:mm:ss}</simple>
            </setHeader>

            <toD uri="pfx-api:fetch?objectType=P&amp;filter=delta-export-products.filter&amp;batchedMode=true&amp;batchSize=5000"/>
            <split>
                <simple>${body}</simple>
                <to uri="pfx-api:fetchIterator"/>
                <to uri="pfx-csv:marshal"/>
                <to uri="file://{{integration.sftp.root}}/export?fileName=products-delta-${date:now:yyyyMMdd}.csv&amp;fileExist=Append"/>
            </split>

            <!-- Save upper bound as the new stored timestamp -->
            <toD uri="pfx-config:set?name={{integration.name}}.${routeId}.export.timestamp&amp;value=${headers.currentExportTimestamp}"/>
        </route>
    </routes>

**Delta filter** (`filters/delta-export-products.filter.xml`):
XML

    <filter id="delta-export-products.filter" sortBy="lastUpdateDate">
        <and>
            <criterion fieldName="lastUpdateDate" operator="greaterThan" value="simple:${headers.lastExportTimestamp}"/>
            <criterion fieldName="lastUpdateDate" operator="lessOrEqual" value="simple:${headers.currentExportTimestamp}"/>
        </and>
    </filter>

The time window (`greaterThan` lower bound + `lessOrEqual` upper bound) ensures records that change **during** the export are picked up in the next run, not lost.

*** ** * ** ***

## Streaming Patterns

### streamCompressedFile

Handle gzip/zip compressed files transparently:
XML

    <to uri="pfx-io:streamCompressedFile"/>

Place this immediately after `<from>` and before CSV unmarshal. Supports `.gz`, `.zip`, and plain files (auto-detected).

### Streaming Unmarshal

For large files, always enable streaming on the split:
XML

    <split aggregationStrategy="recordsCountAggregation" streaming="true">
      <tokenize group="20000" token="
    "/>
      <toD uri="pfx-csv:unmarshal?{{pfx:csv.settings}}&amp;skipHeaderRecord=true"/>
      <toD uri="pfx-api:loaddata?..."/>
      <setBody><constant/></setBody>
    </split>

Without `streaming="true"`, the entire file is loaded into memory before splitting.

### Batch Size Tuning

The `group` attribute on `<tokenize>` controls how many lines are sent per batch:  

|       Object Type       | Recommended Batch Size |          Rationale          |
|-------------------------|------------------------|-----------------------------|
| P, C, SL (master data)  | 20,000                 | Balanced throughput/memory  |
| PX, CX, SX (extensions) | 20,000                 | Same as master              |
| DS/DMDS (data sources)  | 40,000-50,000          | Simple records, high volume |
| PPV (LTV/MLTV2/MLTV3)   | 5,000-10,000           | Heavier records             |

Tune based on environment: reduce for memory-constrained instances, increase for high-throughput.

## onCompletion Chaining

Execute actions after a route completes successfully:
XML

    <route id="import-products">
      <from uri="..."/>

      <onCompletion onCompleteOnly="true">
        <!-- Trigger CFS calculation after successful import -->
        <toD uri="pfx-api:executeCalculation?calculationName={{pfx:cfs.name}}&amp;connection={{pfx:connection}}"/>
      </onCompletion>

      <!-- ... import logic ... -->
    </route>

* `onCompleteOnly="true"` --- only fires on success (not on failure)

* Use for: CFS triggers, DMDS flush, notifications, cleanup

* See [Post-Load CFS/Flush Pattern](patterns/post-load-cfs-flush.md) for detailed examples

## Multicast (Parallel Processing)

Load multiple object types from a single file simultaneously:
XML

    <multicast parallelProcessing="true">
      <to uri="direct:load-products"/>
      <to uri="direct:load-product-extensions"/>
    </multicast>

Use when a single CSV file feeds both P and PX (or C and CX). Each branch gets a copy of the exchange.

*** ** * ** ***

## Common Pitfalls

* **Route ID must match the file name** (without `.xml`). File `import-products.xml` → `id="import-products"`. A mismatched ID causes a deployment failure --- IM cannot start the route.

* Every route must have a unique id across the entire project

* **Always use** `&amp;` **for** `&` **in XML attribute values**

* DMDS imports must include onCompletion with pfx-api:flush

* Use {{property}} placeholders for paths, credentials --- never hardcode

* Use split + tokenize for CSV file imports (batch processing)

* Use batchedMode=true for large data exports

* Route files use a \<routes xmlns="[http://camel.apache.org/schema/spring"\>](http://camel.apache.org/schema/spring%22%3E) wrapper --- no beans wrapper or XML declaration needed

* **PX and CX** have no `extensionName` URI parameter --- the table name is set as `<constant expression="{TableName}" out="name"/>` in the mapper

* **Do not use** `connection=pricefx` --- the default connection is `pricefx` and is used automatically

*** ** * ** ***

## Rules

* Route ID MUST match file name without `.xml` --- mismatched ID causes deployment failure

* Use `<routes xmlns="http://camel.apache.org/schema/spring">` wrapper --- no `<beans>`, no `<?xml>` declaration

* All `&` in URI attributes MUST be `&amp;`

* DMDS imports MUST include `onCompletion` with `pfx-api:flush` (outside the `<split>` block)

* NEVER use `noop=true` on `file:` consumers --- use `{{archive.file}}` + `{{read.lock}}`

* NEVER hardcode file paths or credentials --- use `{{property}}` placeholders

* NEVER include `connection=pricefx` --- it's the default

* PX/CX: table name goes in the mapper as `<constant out="name"/>`, NOT as a URI parameter

* Inside batched export split: use `pfx-api:fetchIterator`, NOT `pfx-api:fetch`

* Quartz cron: spaces replaced with `+` in URI

*** ** * ** ***

## Quick Reference: Common From URIs

|            **URI**             |       **Description**       |
|--------------------------------|-----------------------------|
| `file:{{path}}`                | Read files from directory   |
| `timer://name?repeatCount=1`   | One-time trigger            |
| `scheduler://name?delay=60000` | Periodic trigger            |
| `direct:routeName`             | Internal route call (sync)  |
| `seda:routeName`               | Internal route call (async) |