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

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.

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

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

Source: pipeline/flow_runner/flow_runner.go (line 87) and 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:

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 (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.

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

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

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

Source: 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:

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

Source: 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:

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 (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:

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

Source: proxy/filters/script/javascript/module/processor/processor.go (lines 35-45).

Example JavaScript usage:

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

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 (lines 56-70).

Panic Recovery

Both processors implement deferred panic recovery to ensure stability:

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 (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:

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 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.

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

Data transformation occurs within the FilterFlow.Process method defined in 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}