# faucet-source-xml

[![Crates.io](https://img.shields.io/crates/v/faucet-source-xml.svg)](https://crates.io/crates/faucet-source-xml)
[![Docs.rs](https://docs.rs/faucet-source-xml/badge.svg)](https://docs.rs/faucet-source-xml)
[![MSRV](https://img.shields.io/crates/msrv/faucet-source-xml.svg)](https://github.com/PawanSikawat/faucet-stream/blob/main/rust-toolchain.toml)
[![License](https://img.shields.io/crates/l/faucet-source-xml.svg)](https://github.com/PawanSikawat/faucet-stream#license)

A config-driven **XML / SOAP API source** for the [faucet-stream](https://github.com/PawanSikawat/faucet-stream) ecosystem. It fetches an XML (or SOAP) HTTP endpoint, converts the response to JSON, pulls the repeating record element out by a dot-separated element path, and streams the records page-by-page into any faucet-stream sink — a file, a database, a warehouse, a queue — with one declarative config and no glue code.

Reach for it when the upstream system only speaks XML or SOAP: legacy enterprise web services, RSS/Atom-style feeds, government data portals, or any HTTP endpoint that returns `application/xml`. The parser is event-driven (streaming `quick-xml`), so peak memory stays bounded by `batch_size` even for large documents.

## Feature highlights

- **Automatic XML → JSON conversion** — every response is parsed into a `serde_json::Value` tree; attributes, text nodes, and repeated elements all map to predictable JSON shapes.
- **Element-path record extraction** — `records_element_path` walks a dot-separated path (e.g. `Envelope.Body.GetUsersResponse.Users.User`) to the repeating element and emits one record per match. A single element collapses to one record; a repeated element fans out to many.
- **SOAP friendly** — set `method: POST` and supply a raw SOAP envelope as `body`; attach SOAP-action and content-type headers via `Custom` auth or `query_params`.
- **Two pagination styles** — page-number and offset/limit, each with a built-in loop guard so a misbehaving endpoint can't spin forever; `max_pages` is a hard cap across both.
- **Pluggable authentication** — `none`, `bearer`, `basic`, or fully custom headers — inline or via a shared provider from the CLI `auth:` catalog (OAuth2 token reuse across matrix rows).
- **Retry with backoff** — transient HTTP failures (5xx / connection resets) are retried up to **3 attempts** with exponential backoff (base 500 ms) via the shared `faucet_core::execute_with_retry`.
- **Bounded-memory streaming** — `Source::stream_pages` accumulates matched subtrees and yields a `StreamPage` every `batch_size` records; `batch_size: 0` drains the whole document into one page.
- **Client built once** — the `reqwest` client is constructed in `new()` and reused for every request and every page.

## Installation

```bash
# As a library:
cargo add faucet-source-xml

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-xml
```

Or pull it in through the umbrella crate:

```bash
cargo add faucet-stream --features source-xml
```

`source-xml` is **not** in the CLI default build — enable it explicitly via the feature flag above.

## Quick start

```yaml
# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://api.example.com
      path: /users.xml
      method: GET
      records_element_path: Response.Users.User
  sink:
    type: jsonl
    config:
      path: ./users.jsonl
```

```bash
faucet run pipeline.yaml
```

## Configuration reference

### Core

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `base_url` | string | — *(required)* | Base URL of the API. |
| `path` | string | — *(required)* | Request path appended to `base_url`. |
| `method` | string | `GET` | HTTP method. Use `POST` for SOAP. |
| `body` | string | *(unset)* | Optional request body — typically a raw SOAP envelope for `POST`. |
| `records_element_path` | string | *(unset)* | Dot-separated path to the repeating element in the converted JSON (e.g. `Envelope.Body.GetUsersResponse.Users.User`). When unset, the whole converted document is emitted as one record. |
| `query_params` | map<string,string> | `{}` | Query parameters added to every request (in addition to pagination params). |

### Authentication

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `auth` | `XmlAuth` *(inline `{type,config}`)* or `{ ref: <name> }` | `{ type: none }` | Authentication — see [Authentication](#authentication). |

### Pagination

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `pagination` | `XmlPagination` | *(unset)* | Pagination strategy — see [Pagination](#pagination). When unset, exactly one request is made. |
| `max_pages` | int | *(unset)* | Hard cap on pages fetched, across either pagination style. |

### Batching

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `batch_size` | int | `1000` | Records per emitted `StreamPage`. The event-driven parser buffers matched subtrees and yields whenever the buffer reaches this size. **`0` = no batching**: the document is drained end-to-end and the entire result set is emitted in a single page. Validated against `MAX_BATCH_SIZE` (1,000,000). |

> `headers` exists on the Rust config struct for programmatic use but is `#[serde(skip)]` — it is **not** settable from YAML/JSON. Use `Custom` auth (or `query_params`) to attach request headers from config.

## Authentication

`auth` accepts either an **inline** `{ type, config }` block or a `{ ref: <name> }` pointer to a shared provider declared in the CLI's top-level `auth:` catalog. Inline variants:

| `type` | `config` | Description |
|--------|----------|-------------|
| `none` | *(none)* | No authentication (default). |
| `bearer` | `{ token: <string> }` | `Authorization: Bearer <token>`. |
| `basic` | `{ username, password }` | HTTP Basic authentication. |
| `custom` | `{ headers: { <name>: <value>, … } }` | Arbitrary headers attached to every request — SOAPAction, API keys, content-type overrides, etc. |

```yaml
# Bearer token (read from the environment)
auth:
  type: bearer
  config:
    token: ${env:FEED_TOKEN}
```

```yaml
# HTTP Basic
auth:
  type: basic
  config:
    username: admin
    password: ${env:XML_PASSWORD}
```

```yaml
# Custom headers — e.g. SOAPAction + an API key
auth:
  type: custom
  config:
    headers:
      SOAPAction: "http://example.com/orders/GetOrders"
      X-API-Key: ${env:API_KEY}
```

```yaml
# Shared provider from the top-level auth: catalog (OAuth2 token reused across rows)
auth: { ref: my_idp }
```

## Examples

### REST XML API with page-number pagination

```yaml
version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://api.example.com
      path: /api/products.xml
      records_element_path: Products.Product
      pagination:
        type: PageNumber
        param_name: page
        start_page: 1
        page_size: 50
        page_size_param: per_page
      max_pages: 20
  sink:
    type: jsonl
    config:
      path: ./products.jsonl
```

### SOAP service with Basic auth (POST envelope)

```yaml
version: 1
pipeline:
  source:
    type: xml
    config:
      base_url: https://soap.example.com
      path: /ws
      method: POST
      auth:
        type: basic
        config:
          username: admin
          password: ${env:SOAP_PASSWORD}
      body: |
        <?xml version="1.0"?>
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
          <soapenv:Body>
            <GetOrders xmlns="http://example.com/orders"/>
          </soapenv:Body>
        </soapenv:Envelope>
      records_element_path: Envelope.Body.GetOrdersResponse.Orders.Order
  sink:
    type: postgres
    config:
      connection_url: ${env:DATABASE_URL}
      table: orders
```

### Offset-paginated feed → MongoDB (Bearer auth)

```yaml
version: 1
name: xml_to_mongodb
pipeline:
  source:
    type: xml
    config:
      base_url: https://feeds.example.com
      path: /catalog
      method: GET
      auth:
        type: bearer
        config:
          token: ${env:FEED_TOKEN}
      query_params:
        format: xml
      records_element_path: catalog.products.product
      pagination:
        type: Offset
        offset_param: offset
        limit_param: limit
        limit: 250
  sink:
    type: mongodb
    config:
      connection_uri: mongodb://localhost:27017
      database: warehouse
      collection: catalog_items
      batch_size: 1000
```

Runnable copies of the last two shapes live in [`cli/examples/xml_to_mongodb.yaml`](https://github.com/PawanSikawat/faucet-stream/blob/main/cli/examples/xml_to_mongodb.yaml) and [`cli/examples/xml_to_s3.yaml`](https://github.com/PawanSikawat/faucet-stream/blob/main/cli/examples/xml_to_s3.yaml).

## Pagination

| Style (`type`) | Fields | Stops when |
|----------------|--------|------------|
| `PageNumber` | `param_name`, `start_page`, `page_size` *(optional)*, `page_size_param` *(optional)* | A page returns zero records, or fewer records than `page_size`. |
| `Offset` | `offset_param`, `limit_param`, `limit` | A page returns fewer records than `limit`, or a loop is detected. |

`max_pages` caps the total number of pages for both styles. With no `pagination` block, exactly one request is made.

## Streaming & batching

The source overrides `Source::stream_pages`. The XML response is parsed with an event-driven `quick-xml` reader: only the subtree matching `records_element_path` is materialized, accumulated into a buffer, and yielded as a `StreamPage` whenever the buffer reaches `batch_size`. Memory is bounded at one page regardless of document size. With `batch_size: 0`, the whole document is drained and emitted in a single page — handy for small lookup payloads or for sinks (SQL `COPY`, BigQuery load jobs) that prefer one large request to many small ones.

This is a one-shot fetch source: each run re-requests the endpoint and has no incremental bookmark / resume support. For incremental loads, encode a watermark in `query_params` or the request `body` (e.g. a `since=${now.date}` param) and drive it from a matrix context or `${now.*}` token.

## Config loading & schema introspection

Load from YAML/JSON files, environment variables, or a `.env` file via the helpers in `faucet_core::config`:

```rust
use faucet_core::config::{load_json, load_env_file};
use faucet_source_xml::XmlStreamConfig;

let config: XmlStreamConfig = load_json("config.json")?;
let config: XmlStreamConfig = load_env_file(".env", "XML")?;
```

Inspect the full JSON Schema with:

```bash
faucet schema source xml
```

## Library usage

```rust
use faucet_core::Source;
use faucet_source_xml::{XmlStream, XmlStreamConfig, XmlAuth, XmlPagination};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let config = XmlStreamConfig::new("https://feeds.example.com", "/catalog")
    .auth(XmlAuth::Bearer { token: std::env::var("FEED_TOKEN")? })
    .query_param("format", "xml")
    .records_element_path("catalog.products.product")
    .pagination(XmlPagination::Offset {
        offset_param: "offset".into(),
        limit_param: "limit".into(),
        limit: 250,
    })
    .with_batch_size(500);

let records = XmlStream::new(config).fetch_all().await?;
println!("fetched {} records", records.len());
# Ok(())
# }
```

To wire it into a pipeline, hand the `XmlStream` to `faucet_core::Pipeline` (batch) or `faucet_core::run_stream` (streaming) alongside any `Sink`. For shared-token reuse across many sources, build a `faucet_core::AuthProvider` once and inject it with `XmlStream::with_auth_provider(provider)`.

## How it works

1. `new()` builds the `reqwest` client **once** and stores it on the struct.
2. Each page issues one request (`base_url + path` with `query_params` + the pagination param), retried up to **3 times** with exponential backoff (base 500 ms) on transient failures via `faucet_core::execute_with_retry`. Non-cloneable request bodies are not retried (surfaced as `FaucetError::Source`). When driven by the CLI, a pipeline-level [`resilience:`](https://pawansikawat.github.io/faucet-stream/cookbook/resilience.html) block replaces these built-in retry defaults with one shared policy — XML honors the policy's `max_attempts`, `base`, `max`, `jitter`, and `retry_on` in full.
3. The response is converted XML → JSON and `records_element_path` is walked to find the repeating element.
4. Records are buffered and yielded as `StreamPage`s of `batch_size`; pagination advances until the stop condition or `max_pages`.

## Lineage dataset URI

`<base_url><path>` with any embedded credentials stripped — e.g. `https://soap.example.com/svc`.

## Feature flags

This crate has no optional features of its own. Enable it in the CLI / umbrella crate via the `source-xml` feature.

## Troubleshooting / FAQ

| Symptom | Likely cause & fix |
|---------|--------------------|
| Zero records returned, no error | `records_element_path` doesn't match the converted JSON. Inspect the response (the converter maps attributes and text nodes); fix the dot-path to the *repeating* element, or drop it to emit the whole document as one record. |
| Only one record when you expected many | The path points at a single element rather than the repeated parent's child. Point at the repeating element itself (e.g. `…Users.User`, not `…Users`). |
| `401` / `403` | Auth missing or wrong. Set the right `auth` variant; for SOAP endpoints that gate on `SOAPAction`, add it via `custom` headers. |
| `FaucetError::Source: request is not cloneable for retry` | A streaming/non-cloneable request body can't be retried. Pass the SOAP envelope as a plain `body` string (the default), which is cloneable. |
| Persistent `5xx` after retries | The source retries transient failures **3 times** with backoff before failing. A persistent 5xx is upstream — check the service; raising your own request timeout won't help. |
| Pagination never stops / fetches too much | Set `max_pages` as a hard cap. Confirm `page_size` (PageNumber) or `limit` (Offset) matches what the API actually returns per page so the "fewer than expected" stop condition fires. |
| Headers set in code aren't sent from YAML | `headers` is `#[serde(skip)]` and not configurable from YAML/JSON. Use `custom` auth (or `query_params`) to attach headers from config. |
| Malformed XML / parse error | The body isn't well-formed XML (often an HTML error page returned with a 200). Verify the endpoint and that auth/headers select the XML representation (e.g. `Accept: application/xml`). |

## See also

- [Connector reference](https://pawansikawat.github.io/faucet-stream/reference/connectors.html) — capability matrix.
- [Authentication cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/auth.html) — inline vs shared `auth:` providers.
- [Pagination cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/pagination.html).
- [`faucet-source-rest`](https://crates.io/crates/faucet-source-rest) — the JSON/REST sibling source.
- [`faucet-source-graphql`](https://crates.io/crates/faucet-source-graphql) — GraphQL source.

## License

Licensed under either of [Apache License, Version 2.0](https://www.apache.org/licenses/LICENSE-2.0) or [MIT license](https://opensource.org/licenses/MIT) at your option.
