RMQ Publisher

RMQ Publisher — a pipe for publishing messages to RabbitMQ. Used to send data to external services via a RabbitMQ broker. Typical scenario: sending push notifications through an external delivery service.

Input data

The pipe expects $in.body — the message body to publish:

• If $in.body is a byte array ([]byte), it is used directly
• If $in.body is a string, it is decoded from base64 to []byte

Typically, an Pack pipe is placed before RMQ Publisher, which forms JSON or XML and writes the result to $in.body.

Parameters

Field	Type	Required	Default	Description
connector_shortname	string	Yes	—	String name of the RMQ connector created in External Data Configuration
content_type	string	No	json	Content type: json or xml
delivery_mode	number	No	—	Delivery mode: 1 (transient — message is not persisted to disk) or 2 (persistent — message is saved to the broker disk)
headers	object	No	—	Additional message headers (map[string]string). Supports templating
target	object	Yes	—	Publication target object (see below)

Target object (target)

The pipe supports two target options:

Exchange:

Field	Type	Required	Description
type	string	Yes	Value: exchange
exchange_name	string	Yes	Exchange name in RabbitMQ
exchange_type	string	No (default: direct)	Exchange type: direct, topic, fanout, headers
routing_key	string	No	Routing key

Queue:

Field	Type	Required	Description
type	string	Yes	Value: queue
queue_name	string	Yes	Queue name in RabbitMQ

Possible errors

• $in.body not found (missing Pack pipe before RMQ Publisher)
• $in.body is not a byte array or string (base64)
• Invalid content_type value (only json or xml)
• Connector not found by connector_shortname
• Connector is not an RMQ connector
• RabbitMQ publication error (network error, broker unavailable, etc.)

Examples

Publish to exchange

A Pack pipe should precede this pipe in the chain to form the message body.

{
  "id": 2,
  "type": "rmq_publish",
  "params": {
    "connector_shortname": "ens_push",
    "content_type": "json",
    "delivery_mode": 2,
    "target": {
      "type": "exchange",
      "exchange_name": "fintechiq.push-manager.send",
      "exchange_type": "topic",
      "routing_key": "fintechiq.push-manager.push.by-text.send"
    }
  },
  "outs": {
    "success": 4,
    "error": 100
  }
}

Publish to queue

{
  "id": 2,
  "type": "rmq_publish",
  "params": {
    "connector_shortname": "my_rmq",
    "content_type": "json",
    "delivery_mode": 2,
    "headers": {
      "message_id": "$origin.send_message_id"
    },
    "target": {
      "type": "queue",
      "queue_name": "notifications.inbox"
    }
  },
  "outs": {
    "success": 3,
    "error": 4
  }
}

Migrating from HTTP Request to RMQ Publisher

If the external service accepts messages via RabbitMQ instead of an HTTP API, replace the HTTP Request pipe with RMQ Publisher. The rest of the chain (Pack, Selector, Log) remains unchanged.

What changes

Pipe http_request (id:3) is replaced with rmq_publish using the same id:

{
  "id": 3,
  "type": "rmq_publish",
  "params": {
    "connector_shortname": "cas_rmq",
    "content_type": "json",
    "delivery_mode": 2,
    "target": {
      "type": "exchange",
      "exchange_name": "cas.notifications",
      "exchange_type": "topic",
      "routing_key": "cas.message.send"
    }
  },
  "outs": {
    "success": 10,
    "error": 12
  }
}

Parameters url, method, timeout, success_codes are removed. Instead — connector_shortname and target. The Pack pipe before RMQ Publisher works the same way — it forms $in.body, which is published to RabbitMQ.

Creating an RMQ connector

Before using the pipe, create a connector in External Data Configuration:

Field	Description
Name	Arbitrary name
Short Name	Unique identifier — used in the pipe's connector_shortname
Host	RabbitMQ broker address
Port	Port (default 5672)
Username	Login
Password	Password
Vhost	Virtual host
SSL	SSL and PEM certificates (if needed)

Routing messages with Selector

To direct messages to different queues or exchanges based on event data, use a Selector before multiple RMQ Publisher pipes.

Scheme:

Pack → Selector → RMQ Publisher (priority)
              ↘ RMQ Publisher (normal)

Both pipes can use the same connector — the connection is cached and shared between pipes.

Example: priority and normal notifications

The $template.priority field determines the routing key:

[
  {
    "id": 1,
    "type": "pack",
    "params": {
      "type": "JSON",
      "template": {
        "messageId": "$in.send_message_id",
        "priority": "$template.priority",
        "body": "$template.text",
        "recipient": "$subscription.client_id"
      }
    },
    "outs": {
      "success": 2,
      "error": 99
    }
  },
  {
    "id": 2,
    "type": "selector",
    "params": {
      "outs": [
        {
          "query": "('$template.priority' = 'high')",
          "out": 10
        },
        {
          "query": "('$template.priority' = 'low')",
          "out": 20
        }
      ]
    },
    "outs": {
      "success": 20,
      "error": 99
    }
  },
  {
    "id": 10,
    "type": "rmq_publish",
    "params": {
      "connector_shortname": "cas_rmq",
      "content_type": "json",
      "delivery_mode": 2,
      "target": {
        "type": "exchange",
        "exchange_type": "topic",
        "exchange_name": "cas.notifications",
        "routing_key": "cas.priority.send"
      }
    },
    "outs": {
      "success": 30,
      "error": 99
    }
  },
  {
    "id": 20,
    "type": "rmq_publish",
    "params": {
      "connector_shortname": "cas_rmq",
      "content_type": "json",
      "delivery_mode": 2,
      "target": {
        "type": "exchange",
        "exchange_type": "topic",
        "exchange_name": "cas.notifications",
        "routing_key": "cas.normal.send"
      }
    },
    "outs": {
      "success": 30,
      "error": 99
    }
  },
  {
    "id": 30,
    "type": "log",
    "params": {
      "message": "published successfully"
    }
  },
  {
    "id": 99,
    "type": "log",
    "params": {
      "message": "error: $error.code $error.message",
      "level": "error"
    }
  }
]

Here:

1. Pack (id:1) forms the JSON message body
2. Selector (id:2) checks $template.priority:
   • high → pipe 10 (routing key cas.priority.send)
   • low → pipe 20 (routing key cas.normal.send)
   • no match → outs.success → pipe 20 (fallback)
3. Each RMQ Publisher publishes to its routing key
4. Both converge on pipe 30 (success) or 99 (error)

Selector conditions are checked in order — the first match wins. If no condition matches, the event goes to outs.success, which is convenient for a fallback route. All event data is available in query: $template.*, $subscription.*, $campaign.*, $in.*, $origin.*.

Different brokers for different routes

If queues are on different RabbitMQ brokers, create two connectors and use different connector_shortname values in each RMQ Publisher pipe.