Custom Pipelines service for External enrichment

Overview

The Data Messages sent to Live Objects can be enriched by the Custom Pipelines service.

You can create your own pipeline to perform one or several enrichment steps on specific Data Messages.

Interactive

As described below, a pipeline is composed by a filter section and at least one step section.

Each pipeline targets Data Messages based on its filter configuration, and has a priorityLevel : at most one pipeline can handle a Data Message.

When a Data Message is handled by a pipeline, pipeline id and the status of the execution will be added in the metadata.transformation.pipeline field.

In case of pipeline failure, the original Data Message is stored and an AuditLog message will be sent with failure details.

Pipeline description

pipeline description
{
    "name": "base64 decoding",
    "description": "pipeline to decode base64 encoded message with external transformation",
    "priorityLevel": 10,
    "enabled": true,
    "filter": {
        "connectors": [ "mqtt" ],
        "encodings": [ "base64" ],
        "groupPaths": [ { "path": "/europe", "includeSubPath": true } ],
        "tags": [["PROD"]]
    },
    "steps" : [ {
        "type": "externalTransformation",
        "url": "http://lo-data-transformation.appspot.com:80/base64_decode"
    } ]
}
Field Type Description

name

String

(Mandatory). Name of the pipeline. Max 1000 characters

description

String

(Optional). Description of the pipeline. Max 2000 characters

priorityLevel

Integer

(Mandatory). Used to prioritize pipelines when a Data Message matches with the filter of several pipelines. The pipeline with the lowest priorityLevel value will be selected. In case of equal priorityLevel value, the older pipeline (based on its creation date) will be picked.

enabled

Boolean

(Mandatory). Indicates if the pipeline can apply or not.

filter

Object

(Optional). Define which Data Message can be processed by this pipeline. A null filter means all Data Message can be processed by the pipeline. Criteria in a filter are combined with a AND boolean logic. OR operator is applied between each elements inside filter’s list.

filter.connectors

List<String>

(Optional). If set, then only Data Message sent through one of these connectors (based on its metadata.connector field) will be selected and enter into the pipeline. Possible connectors are: "http", "lora", "mqtt", "sms", "x-connector"

filter.encodings

List<String>

(Optional). If set, then only Data Message with one of these encodings (based on its metadata.encoding field) will be selected and enter into the pipeline.

filter.groupPaths

List<GroupPath>

(Optional). If set, then only Data Message originated from one of these groupPaths (based on its metadata.group field) will be selected and enter into the pipeline.

filter.tags

List<List<String>>

(Optional). If set, then only Data Message with these tags description (based on its tags field) will be selected. There is a match if at least one group of tags is a match. A group of tags is a match if the tags of the message contains all elements of this group. For instance [["HIGH", "ALERT"],["PROD"]] will match any message containing 'PROD' tag; and also any message containing both 'HIGH' and 'ALERT' tags.

steps

List<Steps>

(Mandatory). Define the processing steps of the pipeline (see steps description below).
A dataMessage can only go through one pipeline. (see priorityLevel field for more information). This information is useful when there is overlap between pipeline filters.
A pipeline can have multiple steps. In this case, the steps are executed sequentially

Pipeline Steps description

For now, one type of pipeline steps is available: externalTransformation.

"externalTransformation" step

Interactive

This step will POST a http request with a DataMessage as body toward an external webhook url. Additional headers can be set for each request.

The Data Message format is described in the messages data model section, without the id field as it is not yet stored.

The response to this request must be the transformed (decoded, enriched…​) Data Message using the same format.

Live Objects will then manage this transformed Data Message response:

Behavior Fields

The value of these fields is kept from the original dataMessage. i.e. not overriden

  • streamId

  • metadata

  • extra

  • value.payload

  • created

The value of these fields is replaced by the value contained in the response dataMessage. i.e. overriden Even if 'null'

  • model

  • location

  • tags

  • value (except value.payload)

If not null in the response dataMessage, the value overrides the original field

  • timestamp

Ignored by the pipeline

Any other fields
The important fields that the response can override even if null are: model, location, tags and value.
The http status of the response must be of the 2xx success family in order to be taken into account
In order to ensure that all Data Messages are quickly delivered into other services (Data Store, Alarming, Routing) there is no retry mechanism. Also, only a limited number of Data Messages can wait in a pipeline queue. You need to ensure that your webhook server is available and can hold the Data Messages traffic.
externalTransformation step
{
    "type": "externalTransformation",
    "name": "additional identifier enrichment",
    "url": "http://lo-data-transformation.appspot.com:80/enrich",
    "headers" : [ "x-transform-header" : [ "account-1234" ] ]
}
Field Type Description

type

String

(Mandatory). Must be set to 'externalTransformation'.

name

String

(Optional). Name of the step. It will be pushed in the 'x-orange-lo-pipeline-step-name' http header. Max 1000 characters

url

String

(Mandatory). URL to POST the Data Message. Authorized ports are: 80, 443, 8080, 8443 and 9243.

headers

Map<String, List<String>>

(Optional). If present, these headers will be added in the http POST request.

If the external transformation fails (for ex. because the JSON is badly formatted, server responds http 503…​), Live Objects will store the original Data Message, and an AuditLog message will be logged with failure details.

Example

Implementing a base64 decoder for all dataMessages with 'base64' metadata.decoding:

First, you must deploy your server that will accept POST request and return decoded Data Message.

Then create the following pipeline:

pipeline description
{
    "name": "base64 decoding",
    "priorityLevel": 10,
    "enabled": true,
    "filter": {
        "encodings": [ "base64" ]
    },
    "steps" : [ {
        "type": "externalTransformation",
        "name": "base64 external decoder",
        "url": "http://lo-data-transformation.appspot.com:80/base64_decode"
    } ]
}

The following Data Message is sent by a device:

Data Message sent by a device
{
   "streamId":"device-001-alarm",
   "timestamp":"2019-12-10T13:57:03Z",
   "model":"v1",
   "value":{
      "payload":"bG93IGJhdHRlcnkgYWxhcm0gOiA5JSByZW1haW5pbmc="
   },
   "metadata":{
      "encoding":"base64"
   }
}

This Data Message will arrive into the custom pipeline service and match with the filter of the provisioned pipeline. So, this message will be POST to the url:

POST http request triggered by the DataMessage
POST http://lo-data-transformation.appspot.com:80/base64_decode
'Content-Type': 'application/json'
'x-orange-lo-pipeline-execution-id': '5e9f80ca-6a72-4252-9251-f014b61cf682'
'x-orange-lo-pipeline-step-name': 'base64 external decoder'
{
   "type":"dataMessage",
   "version":1,
   "streamId":"device-001-alarm",
   "timestamp":"2019-12-10T13:57:03Z",
   "model":"v1",
   "value":{
      "payload":"bG93IGJhdHRlcnkgYWxhcm0gOiA5JSByZW1haW5pbmc="
   },
   "metadata":{
      "encoding":"base64"
   }
}

Here is the body response from the remote server (that has performed the payload decoding and has added a tag):

POST http response 200 OK
{
   "model":"v1_base64_decoded",
   "value":{
      "alarm":"low battery alarm : 9% remaining",
      "battery_level":9
   },
   "tags":[
      "ALARM",
      "BASE64_DECODED"
   ]
}

Then the stored Data Message will be:

Stored Data Message
{
   "streamId":"device-001-alarm",
   "timestamp":"2019-12-10T13:57:03Z",
   "model":"v1_base64_decoded",
   "value":{
      "payload":"bG93IGJhdHRlcnkgYWxhcm0gOiA5JSByZW1haW5pbmc=",
      "alarm":"low battery alarm : 9% remaining",
      "battery_level":9
   },
   "tags":[
      "ALARM",
      "BASE64_DECODED"
   ],
   "metadata":{
      "encoding":"base64",
      "transformation":{
         "pipeline":{
            "id":"a07767f6-809e-4943-8b79-5efb96bd1535",
            "success":true
         }
      }
   }
}
Version 2026.7.0 • Updated date 2026-08-11 12:37:29 UTC