# How the Pipeline Processor in INFINI Gateway Manages Data Transformation and Filtering

> Discover how INFINI Gateway pipelines process messages, apply filters, and commit offsets for efficient data transformation and filtering.

- Repository: [INFINI Labs/gateway](https://github.com/infinilabs/gateway)
- Tags: deep-dive
- Published: 2026-03-04

---

**The pipeline processor in INFINI Gateway pulls messages from queues, executes a configurable chain of filter plugins against a `fasthttp.RequestCtx`, and handles offset commits to enable robust data transformation and filtering workflows.**

The INFINI Gateway pipeline processor serves as the core execution engine for data transformation and filtering within the open-source gateway. It treats every incoming request as a pipeline job, processing raw messages through configurable flows to mutate headers, query parameters, and body content before forwarding to downstream services.

## Processor Registration and Initialization

All pipeline processors register with the core framework at `init()` time using `pipeline.RegisterProcessorPlugin`. This registration makes the processors discoverable by the pipeline builder when parsing configuration files.

```go
func init() {
    pipeline.RegisterProcessorPlugin("flow_runner", New)
}

```

```go
func init() {
    pipeline.RegisterProcessorPlugin("fast_flow_runner", New)
}

```

*Source:* [`pipeline/flow_runner/flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/pipeline/flow_runner/flow_runner.go) (line 87) and [`pipeline/fast_flow_runner/flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/pipeline/fast_flow_runner/flow_runner.go) (line 80).

The framework provides two variants: **`flow_runner`** for standard queue consumption and **`fast_flow_runner`** for low-latency processing.

## Data Ingestion from Message Queues

The pipeline processor manages data transformation by first pulling raw messages from configured queues. Both processors follow the same ingestion pattern:

1. Resolve queue configuration via `queue.GetOrInitConfig`
2. Acquire a consumer instance (`queue.AcquireConsumer`) — standard in `flow_runner`
3. Fetch messages using `FetchMessages` or `PopTimeout` (fast variant)
4. Decode the raw payload into a `fasthttp.RequestCtx` using `ctx.Request.Decode`

*Excerpt from [`flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/flow_runner.go):*

```go
messages, timeout, err := consumerInstance.FetchMessages(ctx1, processor.config.Consumer.FetchMaxMessages)
...
ctx := acquireCtx()
err = ctx.Request.Decode(pop.Data)

```

*Source:* [`pipeline/flow_runner/flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/pipeline/flow_runner/flow_runner.go) (lines 26-28, 37-44).

## Flow Lookup and Filter Chain Execution

Once the request context is prepared, the pipeline processor looks up the configured flow using `GetFlowProcess`. This function returns a closure that executes the filter chain.

```go
flowProcessor := common.GetFlowProcess(processor.config.FlowName)

```

The lookup implementation retrieves the flow from an internal cache or builds it from configuration if missing:

```go
func GetFlowProcess(flowID string) func(ctx *fasthttp.RequestCtx) {
    flow := MustGetFlow(flowID)
    return flow.Process
}

```

*Source:* [`common/flow.go`](https://github.com/infinilabs/gateway/blob/main/common/flow.go) (lines 27-30).

### The Filter Chain Loop

The `FilterFlow.Process` method iterates over the slice of `pipeline.Filter` objects, executing each transformation in sequence:

```go
for _, v := range flow.Filters {
    if !ctx.ShouldContinue() { break }
    ctx.AddFlowProcess(v.Name())
    v.Filter(ctx)
}

```

*Source:* [`common/flow.go`](https://github.com/infinilabs/gateway/blob/main/common/flow.go) (lines 64-81).

Each filter mutates the `fasthttp.RequestCtx` in place, modifying headers, query arguments, or body content before passing control to the next filter.

## Built-in Transformation Filters

The pipeline processor supports extensive data transformation through built-in filter plugins. All filters implement the `pipeline.Filter` interface:

```go
type Filter interface {
    Name() string
    Filter(ctx *fasthttp.RequestCtx)
}

```

Common transformation filters include:

- **`set_request_header`** — Replaces or adds request headers
- **`set_request_query_args`** — Mutates query string parameters
- **`set_response_header`** — Alters response headers
- **`request_body_json_set`** / **`request_body_json_del`** — JSON manipulation on request bodies
- **`response_body_regex_replace`** — Regex-based body rewriting

*Source:* [`proxy/filters/transform/set_header.go`](https://github.com/infinilabs/gateway/blob/main/proxy/filters/transform/set_header.go) (lines 21-96) and related transform files.

## JavaScript Extensibility

Beyond Go-based filters, the pipeline processor supports JavaScript-based transformations through a native module system. The gateway registers a `processor` module that exposes Beat-style processors to JavaScript:

```go
// Registration of native processors for JavaScript
func init() {
    // processor module registration
}

```

*Source:* [`proxy/filters/script/javascript/module/processor/processor.go`](https://github.com/infinilabs/gateway/blob/main/proxy/filters/script/javascript/module/processor/processor.go) (lines 35-45).

Example JavaScript usage:

```js
var p = new processor.Dissect({ tokenizer: "%{key}: %{value}" });
var chain = new processor.Chain().Add(p);

```

## Offset Management and Error Handling

The pipeline processor manages message offsets through configurable commit strategies. After processing each message, the runner updates the queue offset based on configuration:

- **After every message** — Immediate commit
- **On idle timeout** — Commit when no new messages arrive
- **When a specific tag appears** (`CommitOnTag`) — Conditional commit based on context tags

```go
if processor.config.CommitOnTag != "" {
    tags, ok := ctx.GetTags()
    if ok && tags[processor.config.CommitOnTag] {
        // commit offset
    }
}
offset = pop.NextOffset

```

*Source:* [`pipeline/flow_runner/flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/pipeline/flow_runner/flow_runner.go) (lines 56-70).

### Panic Recovery

Both processors implement deferred panic recovery to ensure stability:

```go
if r := recover(); r != nil {
    log.Errorf("error in flow_runner [%v], [%v]", processor.config.FlowName, v)
    ctx.RecordError(fmt.Errorf("flow runner panic: %v", r))
}

```

*Source:* [`pipeline/flow_runner/flow_runner.go`](https://github.com/infinilabs/gateway/blob/main/pipeline/flow_runner/flow_runner.go) (lines 51-70).

A dedicated `Stop()` method signals the processing loop to shut down gracefully via `signalChannel`.

## Configuration Example

The following YAML configuration demonstrates a complete pipeline processor setup with data transformation and filtering:

```yaml
processor:
  type: flow_runner
  flow: my_transform_flow
  input_queue: my_queue
  skip_empty_queue: true

flow:
  id: my_transform_flow
  filters:
    - type: set_request_header
      headers:
        - "X-User-Id->{{user.id}}"
        - "X-Trace-Id->{{trace.id}}"
    - type: request_body_json_set
      fields:
        - "metadata->{{metadata}}"
    - type: set_response_header
      headers:
        - "Cache-Control->no-store"

```

**Processing sequence:**

1. `flow_runner` pulls a message from `my_queue`
2. Creates a `RequestCtx` and calls `GetFlowProcess("my_transform_flow")`
3. Executes three filters in order: header injection, JSON body modification, and response header setting
4. Commits the offset and continues to the next message

## Summary

- The **pipeline processor** in INFINI Gateway acts as a high-performance worker that consumes messages from queues and executes configurable transformation flows.
- **Two processor variants** exist: `flow_runner` for standard processing and `fast_flow_runner` for low-latency scenarios.
- **Filter chains** defined in [`common/flow.go`](https://github.com/infinilabs/gateway/blob/main/common/flow.go) process each request through an ordered sequence of `pipeline.Filter` implementations that mutate `fasthttp.RequestCtx`.
- **Built-in filters** handle headers, query arguments, JSON body manipulation, and regex replacements, while **JavaScript modules** enable custom scripting via the `processor` native module.
- **Robust error handling** includes panic recovery, configurable offset committing (per-message, on-tag, or timeout-based), and graceful shutdown via signal channels.

## Frequently Asked Questions

### What is the difference between flow_runner and fast_flow_runner in INFINI Gateway?

The `flow_runner` processor uses `queue.AcquireConsumer` and `FetchMessages` for standard queue consumption with configurable batch sizes, while `fast_flow_runner` utilizes `queue.PopTimeout` for single-message, low-latency processing. Both implement the same `FilterFlow.Process` chain but optimize for different throughput requirements.

### How does the pipeline processor handle message failures and ensure data integrity?

The processor implements deferred panic recovery in the processing loop to catch runtime errors without crashing the worker. It records errors via `ctx.RecordError` or `ctx.Failed` and supports configurable offset committing strategies—including commit-on-tag and idle timeout—to ensure messages are only acknowledged after successful processing or specific business logic conditions are met.

### Can I extend the pipeline processor with custom transformation logic?

Yes, the pipeline processor supports extension through two mechanisms: implementing the `pipeline.Filter` interface in Go to create native plugins that mutate `fasthttp.RequestCtx`, or using the JavaScript processor module which exposes Beat-style processors via the `processor` native module registered in [`proxy/filters/script/javascript/module/processor/processor.go`](https://github.com/infinilabs/gateway/blob/main/proxy/filters/script/javascript/module/processor/processor.go).

### Where does the actual data transformation occur in the pipeline processor?

Data transformation occurs within the `FilterFlow.Process` method defined in [`common/flow.go`](https://github.com/infinilabs/gateway/blob/main/common/flow.go) (lines 64-81), which iterates over the slice of `pipeline.Filter` objects. Each filter—such as `set_request_header` or `request_body_json_set`—mutates the `fasthttp.RequestCtx` in place before passing control to the next filter in the chain.