Pipelines

Pipeline is a chain of data processing elements that describes how a channel interacts with external services. A pipeline is defined as a JSON array and configured on the Pipelines tab when setting up a channel.

Pipe is an element of a pipeline that performs a single specific action: sends an HTTP request, packs data into JSON, writes a log, registers an event, etc. Each pipe has an input and can have one or more outputs. If a pipe has no outputs, the event exits the pipeline.

A complete list of pipes with parameter descriptions and examples is available in the Pipes section.

How a pipeline works

A pipeline works by sequentially passing an event from pipe to pipe:

• Each pipe has a unique id, which also serves as its input.
• The pipe's outputs (outs) specify the id of the next pipe that the event will move to on success (success) or error (error).
• If an output is not specified, the event exits the pipeline.
• Data passed to a pipe from the previous pipe is accessible via the $in keyword.
• The original pipeline data is accessible via $origin.

You can access field values inside a pipe using the templating language.

Flow visualization

[Pipe 1: Pack] ──success──> [Pipe 2: HTTP Request] ──success──> [Pipe 3: Eventer (send)]
                                    │
                                   error
                                    v
                              [Pipe 4: Eventer (undeliv)]

Each arrow is defined through the pipe's outs: success specifies the next pipe's id on success, error — on error. If an output is not specified, the event exits the pipeline.

Working with form data

Data from channel entity fields (Template, Campaign, Sender, Resource, Subscription, Account) linked to an External Object with the List operation enters the pipeline as an array. This is because the user can select multiple values (for example, multiple profiles or segments).

If you need to extract a specific value from the array, use indexing via .%<index> (zero-based indexing):

$template.profile.%0.id

This returns the id of the first selected item. To access the second — %1, the third — %2, and so on.

If you need to access the entire array, use the keyword without indexing:

$template.profile

To unpack an array into a repeating structure (for example, when building JSON with multiple attachments), use the $$ prefix. For more details, see the Templating Language section.

Pipe structure

Each pipe is described as a JSON object with the following fields:

Field	Example	Required	Description
id	5	Yes	Unique identifier of the pipe in the pipeline. Also serves as its input.
type	"log"	Yes	Pipe type (determines which action is performed).
params	{}	No	Parameters specific to this pipe type.
outs	OutsObject	No	Pipe outputs — determine which pipe the event will move to.

Pipe outputs (OutsObject)

Field	Example	Required	Description
success	3	No	id of the pipe the event will move to on successful execution.
error	4	No	id of the pipe the event will move to on error.

Workers

Each pipe in a pipeline is represented as an array of workers. The number of workers is configured in the platform configuration file via the PIPE_WORKER_SIZE parameter.

Pipeline example

Below is an example of a pipeline that builds a JSON request, sends it over HTTP, and depending on the result, registers a delivery or non-delivery event:

[
  {
    "id": 1,
    "type": "pack",
    "params": {
      "type": "JSON"
    },
    "outs": {
      "success": 2
    }
  },
  {
    "id": 2,
    "type": "http_request",
    "params": {
      "timeout": 10,
      "content_type": "json",
      "success_codes": [200, 202]
    },
    "outs": {
      "success": 3,
      "error": 4
    }
  },
  {
    "id": 3,
    "type": "event",
    "params": {
      "name": "mydeliv",
      "event": {
        "send_message_id": "$origin.send_message_id"
      }
    }
  },
  {
    "id": 4,
    "type": "event",
    "params": {
      "name": "myundeliv",
      "event": {
        "send_message_id": "$origin.send_message_id",
        "message": "$error.code $error.message"
      }
    }
  }
]

Here:

1. Pipe 1 (Pack) packs data into JSON and passes the event to pipe 2.
2. Pipe 2 (HTTP Request) sends an HTTP request. On success, the event goes to pipe 3; on error, to pipe 4.
3. Pipe 3 (event) registers a delivery event.
4. Pipe 4 (event) registers a non-delivery event with the error code and message.