Transformação

É possível aplicar transformações nas mensagens trafegadas no fluxo. Pode ser aplicada em qualquer tipo de configuração, inclusive com uma configuração de Processamento em Streaming.

No momento, as transformações funcionam apenas com payloads com Content-Type application/json.

Tipos

As seguintes transformações são aceitas:

HTTP

Permite que um endpoint HTTP externo seja consumido durante o processamento e atribui o resultado à mensagem.

Bloblang

Permite um mapeamento usando a sintaxe bloblang e atribui o resultado à mensagem.

Para realizar testes da sintaxe a ferramenta Bloblang Walkthrough pode ser utilizada.

Delete

Permite uma a sintaxe bloblang que retorne um valor booleano indicando se a mensagem deve ser excluída.

Parâmetros

Parâmetros aceitos na configuração de transformação:

Campo	Descrição	Valor
transformer.description	Descrição do estágio de transformação	Tipo: string
transformer.workflow	Configuração de prioridade de execução do fluxo	Tipo: [][]string Valor padrão: [[id1,id2,...]]
transformer.steps	Lista com as etapas de transformação	Tipo: list Obrigatório
transformer.steps.type	Tipo de transformação	Tipo: enum[BLOBLANG HTTP DELETE] Obrigatório
transformer.steps.id	Id da transformação. Campo que recebe o resultado da transformação se o campo override é falso.	Tipo: string [alphanum_-] Obrigatório - Único entre as etapas
transformer.steps.ignore_error	Ignorar erro durante a transformação	Tipo: boolean Valor padrão: false
transformer.steps.override	Substituir conteúdo da mensagem com o resultado da transformação	Tipo: boolean Valor padrão: false
transformer.steps.http	Transformação do tipo Http	Tipo: object Obrigatório se o campo type for igual a HTTP
transformer.steps.http.url	Url do endpoint http a ser chamado	Tipo: string Obrigatório se dynamic_url não for informado
transformer.steps.http.dynamic_url.mapping	Url do endpoint http a ser chamado na sintaxe bloblang permitindo que sejam adicionados parâmetros dinâmicos a partir do payload da mensagem. Ex: \"https://example.com?param1=%s&param2=%s\".format(field1,parent.field2)	Tipo: string Obrigatório se url não for informado
transformer.steps.http.verb	Tipo da requisição	Tipo: enum[GET POST] Obrigatório
transformer.steps.http.headers	Cabeçalhos da requisição	Tipo: object Valor padrão: -
transformer.steps.http.timeout_seconds	Tempo máximo da resposta em segundos	Tipo: int [1-60] Valor padrão: 5
transformer.steps.http.ignore_error_codes	Códigos de erro da reposta que devem ser ignorados	Tipo: []int Valor padrão: []
transformer.steps.http.body.mapping	Body da requisição na sintaxe bloblang	Tipo: string Valor padrão: "root = this"
transformer.steps.http.rate_limit	Configuração de limites da requisição	Tipo: object Valor padrão: {}
transformer.steps.http.rate_limit.count	Quantidade máxima de requisições	Tipo: int Valor padrão: -
transformer.steps.http.rate_limit.interval_seconds	Intervalo em segundos	Tipo: int Valor padrão: -
transformer.steps.http.basic_auth	Autenticação básica da requisição	Tipo: object Valor padrão: {}
transformer.steps.http.basic_auth.user_id	Identificador de acesso	Tipo: string Obrigatório
transformer.steps.http.basic_auth.password	Senha de acesso	Tipo: string Obrigatório
transformer.steps.http.oauth	Autenticação oauth da requisição	Tipo: object Valor padrão: {}
transformer.steps.http.oauth.consumer_key	Chave de consumo da API	Tipo: string Obrigatório
transformer.steps.http.oauth.consumer_secret	Segredo de consumo da API	Tipo: string Obrigatório
transformer.steps.http.oauth.access_token	Token de acesso da API	Tipo: string Obrigatório
transformer.steps.http.oauth.access_token_secret	Segredo do token de acesso da API	Tipo: string Obrigatório
transformer.steps.http.oauth2	Autenticação oauth 2 da requisição	Tipo: object Valor padrão: {}
transformer.steps.http.oauth2.client_key	Chave do cliente da API	Tipo: string Obrigatório
transformer.steps.http.oauth2.client_secret	Segredo do cliente da API	Tipo: string Obrigatório
transformer.steps.http.oauth2.token_url	Url do serviço de autenticação	Tipo: string Obrigatório
transformer.steps.http.oauth2.scopes	Escopos de autenticação	Tipo: []string Valor padrão: []
transformer.steps.http.oauth2.auth_style	Localização dos parâmetros com a autenticação	Tipo: enum[param header body] Valor padrão: param
transformer.steps.http.oauth2.endpoint_params	Parâmetros adicionais do serviço de autenticação	Tipo: map[string][]string Valor padrão: {}
transformer.steps.bloblang	Transformação do tipo Bloblang	Tipo: object Obrigatório se o campo type for igual a BLOBLANG
transformer.steps.bloblang.mapping	Sintaxe bloblang com a transformação a ser aplicada na mensagem	Tipo: string Obrigatório
transformer.steps.delete	Transformação do tipo Delete	Tipo: object Obrigatório se o campo type for igual a DELETE
transformer.steps.delete.mapping	Sintaxe bloblang que resulte em um valor booleano indicando que a mensagem deve ser removida	Tipo: string Obrigatório

Requisição

Para habilitar a transformação a seguinte requisição pode ser realizada:

curl --location '<HOST>/v1/configs' \
--header 'Content-Type: application/json' \
--header 'Authorization: <ACCESS-TOKEN>' \
--data '{
{
  "transformer": {
    "description": "TRANSFORMER-DESCRIPTION",
    "workflow": [
      [
        "http"
      ],
      [
        "bloblang"
      ],
      [
        "delete"
      ]
    ],
    "steps": [
      {
        "id": "http",
        "type": "HTTP",
        "ignore_error": false,
        "override": false,
        "http": {
          "url": "https://example.com",
          "dynamic_url": {
            "mapping": "\"https://example.com?param1=%s&param2=%s\".format(field1,parent.field2)"
          },
          "verb": "GET",
          "headers": {
            "key": "value"
          },
          "timeout_seconds": 10,
          "ignore_error_codes": [
            1,
            2,
            3
          ],
          "body": {
            "mapping": "root = this"
          },
          "rate_limit": {
            "count": 100,
            "interval_seconds": 10
          },
          "basic_auth": {
            "user_id": "USERNAME",
            "password": "PASSWORD"
          },
          "oauth": {
            "consumer_key": "CONSUMER_KEY",
            "consumer_secret": "CONSUMER_SECRET",
            "access_token": "ACCESS_TOKEN",
            "access_token_secret": "ACCESS_TOKEN_SECRET"
          },
          "oauth2": {
            "client_key": "CLIENT_KEY",
            "client_secret": "CLIENT_SECRET",
            "token_url": "TOKEN_URL",
            "auth_style": "body",
            "scopes": [
              "SCOPE"
            ],
            "endpoint_params": {
              "key": [
                "value"
              ]
            }
          }
        }
      },
      {
        "id": "bloblang",
        "type": "BLOBLANG",
        "ignore_error": false,
        "override": false,
        "bloblang": {
          "mapping": "root = this.http"
        }
      },
      {
        "id": "delete",
        "type": "DELETE",
        "ignore_error": false,
        "override": false,
        "delete": {
          "mapping": "root = this.bloblang == \"OK\""
        }
      }
    ]
  }
}
}'

No exemplo acima foi gerada uma configuração do Firehose que aplica 3 etapas de transformação de forma sequencial na mensagem.

A primeira etapa realiza uma requisição http e armazena o resultado no campo http. A segunda etapa recupera o resultado do campo adicionado pela etapa anterior e armazena o resultado no campo bloblang. A última etapa verifica se o valor do campo adicionado na etapa anterior é igual à OK, em caso positivo a mensagem será apagada.