Connections Overview

Connections let you integrate Ditto with external messaging systems so that devices can exchange data with their digital twins through protocols like AMQP, MQTT, HTTP, and Kafka.

Overview

You integrate your Ditto instance with external messaging services – such as Eclipse Hono, RabbitMQ, Apache Kafka, or any HTTP endpoint – by creating connections.

A connection represents a communication channel that uses a transport protocol to transmit Ditto Protocol messages. Ditto supports one-way and two-way communication, enabling consumer/producer scenarios as well as full command-and-response workflows.

All connections are configured and supervised by Ditto’s Connectivity service.

To create and manage connections, use the HTTP API or DevOps piggyback commands.

Connection model

The following schema defines the connection model:

Supported connection types

Ditto supports these connection types:

• AMQP 0.9.1
• AMQP 1.0
• MQTT 3.1.1
• MQTT 5
• HTTP 1.1
• Kafka 2.x

The format of sources and targets addresses depends on the connectionType and is documented in each protocol binding page.

Sources

Sources consume messages from external systems. Inbound messages can be:

• Commands
• Messages
• Live commands/responses/events
• Acknowledgements

A source contains:

• addresses – interpreted as queues, topics, etc. depending on the connection type
• consumerCount – how many consumers attach to each address
• authorizationContext – authorization subjects used to authorize inbound messages (see Authorization)
• enforcement – filters to verify that a device only modifies its own digital twin
• acknowledgementRequests – controls QoS 1 processing
• declaredAcks – labels of acknowledgements this source may send
• headerMapping – maps external headers to internal headers (see Header mapping)
• replyTarget – where to publish responses to incoming commands

Source enforcement

By default, Ditto does not verify whether the device identity in an inbound message matches the targeted thing. You can add enforcement to ensure a device only modifies its own digital twin.

Enforcement requires that the external system provides a verified device identity (for example, in a message header).

The enforcement configuration has two fields:

• input – where the device identity is extracted from
• filters – patterns matched against the input; at least one must match or the message is rejected

Placeholders for input:

Placeholders for filters:

Example: Device sensor:temperature1 provides its identity in a device_id header. To enforce that it can only write to its own twin:

{
  "addresses": ["telemetry/hono_tenant"],
  "authorizationContext": ["ditto:inbound-auth-subject"],
  "enforcement": {
    "input": "{{ header:device_id }}",
    "filters": ["{{ thing:id }}"]
  }
}

Source acknowledgement requests

To process inbound messages with “at least once” (QoS 1) semantics instead of the default “at most once” (QoS 0), configure acknowledgementRequests/includes to request the “twin-persisted” acknowledgement. The message is then technically acknowledged only after the twin is successfully persisted.

The optional filter field uses an fn:filter() expression to control when acknowledgements are requested:

{
  "addresses": ["<source>"],
  "authorizationContext": ["ditto:inbound-auth-subject"],
  "headerMapping": {
    "qos": "{{ header:qos }}"
  },
  "acknowledgementRequests": {
    "includes": ["twin-persisted", "{{connection:id}}:my-custom-ack"],
    "filter": "fn:filter(header:qos,'ne','0')"
  }
}

Source declared acknowledgement labels

Acknowledgements sent via a source must have their labels declared in the declaredAcks array. Labels must be prefixed by the connection ID (or {{connection:id}}) followed by a colon:

{
  "addresses": ["<source>"],
  "authorizationContext": ["ditto:inbound-auth-subject"],
  "declaredAcks": [
    "{{connection:id}}:my-custom-ack"
  ]
}

Source header mapping

You can apply an optional header mapping to inbound messages. Mapped headers are added to the Ditto protocol message produced by payload mapping:

{
  "addresses": ["<source>"],
  "authorizationContext": ["ditto:inbound-auth-subject"],
  "headerMapping": {
    "correlation-id": "{{ header:message-id }}",
    "content-type": "{{ header:content-type }}"
  }
}

Source reply target

A source may define a reply target to publish responses to incoming commands. The reply target’s address and header mapping are defined within the reply target, while its payload mapping is inherited from the parent source.

To publish responses at the address from the incoming command’s reply-to header, configure source header mapping and reply target together. If an incoming command lacks the reply-to header, no response is published:

{
  "headerMapping": {
    "reply-to": "{{ header:reply-to }}"
  },
  "replyTarget": {
    "enabled": true,
    "address": "{{ header:reply-to }}",
    "headerMapping": {
      "correlation-id": "{{ header:correlation-id }}"
    },
    "expectedResponseTypes": ["response", "error", "nack"]
  }
}

The expectedResponseTypes control which responses are published:

• response – successful responses and positive acknowledgements
• error – error responses
• nack – negative acknowledgements

Response diversion

Sources can redirect responses to different connections instead of the configured reply target using special header mapping keys. See Response diversion for details.

Static response diversion – redirect all responses to a fixed connection:

{
  "addresses": ["commands/sensor"],
  "authorizationContext": ["ditto:sensor-commands"],
  "headerMapping": {
    "divert-response-to-connection": "analytics-connection",
    "divert-expected-response-types": "response,error"
  },
  "replyTarget": {
    "enabled": true,
    "address": "responses/sensor"
  }
}

Where:

• divert-response-to-connection: Target connection ID for diversion
• divert-expected-response-types: Comma-separated list of response types to divert

Dynamic response diversion – route based on message content using a JavaScript payload mapper:

{
  "addresses": ["commands/+"],
  "authorizationContext": ["ditto:device-commands"],
  "headerMapping": {
    "divert-expected-response-types": "response,error"
  },
  "payloadMapping": ["response-router"]
}

function mapToDittoProtocolMsg(headers, textPayload, bytePayload, contentType) {
  var parsedPayload = JSON.parse(textPayload);
  var dittoHeaders = {
    "correlation-id": headers["correlation-id"],
    "divert-response-to": determineTargetConnection(headers, parsedPayload),
    "divert-expected-response-types": "response,error"
  };
  return Ditto.buildDittoProtocolMsg(
    namespace, name, group, channel, criterion, action, path,
    dittoHeaders, value
  );
}

function determineTargetConnection(headers, payload) {
  if (payload.priority === "high") {
    return "priority-processing-connection";
  } else if (payload.deviceType === "sensor") {
    return "sensor-analytics-connection";
  } else {
    return "default-processing-connection";
  }
}

Targets

Targets publish messages to external systems. Outbound messages can be:

• Thing events
• Thing messages
• Live commands/responses/events
• Policy announcements
• Connection announcements

A target contains:

• address – interpreted as a queue, topic, etc. depending on the connection type
• topics – which message types to publish
• authorizationContext – authorization subjects that must have READ permission
• headerMapping – maps Ditto protocol headers to external headers

Target topics and filtering

You define which message types to publish via the topics array. You can filter by namespaces, RQL expressions and placeholder pipelines:

Filter parameters use HTTP query parameter syntax (? for the first, & for subsequent). A topic may carry one filter parameter holding an RQL expression and any number of fn-filter parameters, each holding a placeholder pipeline expression (see filtering with placeholder functions below). The parameter name tells the two apart – filter is always RQL, fn-filter is always a placeholder pipeline. All given filters must match for a signal to be published (AND semantics).

Write the filter expressions as shown in the examples: |, ', ", ,, (, ), :, *, ? and = need no URL-encoding. Parameter values are URL-decoded every time a topic string is parsed (e.g. %7C becomes a literal |), but they are not re-encoded when the connection is stored, so a stored topic is decoded once more whenever the connection is loaded again. Three characters can therefore not be used at all in the filter values of a connection target topic – neither in an RQL value or like pattern nor in the compared value of an fn-filter:

• & – a raw & ends the parameter. %26 is decoded to & on the first parse, so the connection is accepted, but that & cuts the expression off when the stored topic is parsed again; an fn-filter truncated that way suppresses every signal. (For WebSocket and SSE filters, which are not stored, %26 works.)
• + – is decoded to a space. %2B is decoded to + on the first parse and to a space on the next one.
• % – %25 is decoded to % on the first parse. When the stored topic is parsed again, that % either fails to decode or is silently decoded together with the two characters following it.

Example:

{
  "address": "<target-address>",
  "topics": [
    "_/_/things/twin/events?namespaces=org.eclipse.ditto&filter=gt(attributes/counter,42)",
    "_/_/things/twin/events?extraFields=attributes/placement&filter=gt(attributes/placement,'Kitchen')",
    "_/_/things/live/messages?namespaces=org.eclipse.ditto",
    "_/_/things/live/commands?fn-filter=resource:path|fn:filter('like','/attributes/*')"
  ],
  "authorizationContext": ["ditto:outbound-auth-subject"]
}

If a target’s topics array lists several topic entries, they are evaluated independently and combined with OR semantics – a signal is published (once) as soon as it matches any one listed topic (each with its own namespaces, filter and fn-filter parameters).

Filtering with placeholder functions

Since Ditto 4.0.0, a target topic may carry fn-filter parameters in addition to (or instead of) the RQL filter parameter, each holding a placeholder pipeline: a placeholder, optionally passed through value-transforming functions of the function library, whose value is finally tested by an fn:filter() stage. An fn-filter is a generic filter on any placeholder which is available for an outbound signal – header:..., topic:..., thing:..., entity:..., feature:..., resource:..., request:..., time:..., connection:id and, for thing events, thing-json:... – see connection target topic filter placeholders for the full list.

Such a pipeline is evaluated per outbound signal against what the signal itself carries – its headers, topic, entity ID, resource, time and, for thing events, the event’s own data – and never against enriched (extraFields) data. An RQL filter matches against thing data plus the topic:, resource: and time: placeholders; an fn-filter needs no thing data and therefore works the same for every signal type a target topic can carry – twin events, live events, live messages and live commands –, also for _/_/things/live/commands, for which an RQL filter is not supported (marked ❌ for “RQL filter” in the table above).

The shape of an fn-filter

An fn-filter always has the same shape – the placeholder whose value is filtered, optional value-transforming fn: stages, and exactly one fn:filter as the last stage:

<placeholder>[|fn:<value stage>...]|fn:filter('<rqlFunction>',<comparedValue>)

fn-filter=topic:action|fn:filter('ne','deleted')
fn-filter=resource:path|fn:filter('like','/features/*')
fn-filter=thing:name|fn:substring-before('-')|fn:filter('eq','sensor')
fn-filter=header:content-type|fn:lower()|fn:filter('like','application/json*')
fn-filter=header:ditto-originator|fn:filter('exists','true')

From top to bottom, these publish: everything but deletions; only signals whose resource path starts with /features/; only signals of things whose name starts with sensor-; only signals whose content-type header starts with application/json, in whatever case it is written; only signals which carry a ditto-originator header.

• <rqlFunction> is one of the RQL functions eq, ne, like and exists, given as a quoted constant.
• <comparedValue> is a quoted constant or another placeholder (e.g. fn:filter('eq',header:expected)); for exists it must be the constant 'true'.
• A quoted constant may be given in single or double quotes and may contain | and ,. To compare with a value containing a single quote, use double quotes (header:x|fn:filter("ne","it's")) and vice versa – a backslash-escaped quote is accepted, but the backslash stays part of the compared value.
• fn:filter always filters the value of the pipeline in front of it – the value to filter is never passed to fn:filter as a parameter.

The publish decision of an fn-filter is binary:

• the pipeline resolves to a value – the target topic is published
• the pipeline stays unresolved – the target topic is suppressed

Think of fn:filter as the stage that returns the boolean publish decision: it keeps the pipeline resolved (publish) when its condition holds and leaves it unresolved (suppress) otherwise.

Each topic of a target can filter on what is relevant for its signal type:

{
  "address": "<target-address>",
  "topics": [
    "_/_/things/twin/events?fn-filter=topic:action|fn:filter('ne','deleted')",
    "_/_/things/live/messages?fn-filter=topic:subject|fn:filter('like','alarm.*')",
    "_/_/things/live/commands?fn-filter=header:ditto-originator|fn:filter('ne','some:excluded-subject')"
  ],
  "authorizationContext": ["ditto:outbound-auth-subject"]
}

• The twin events are published unless they are deletions: topic:action resolves to deleted for e.g. a deleted thing or attribute.
• Of the live messages only those are published whose subject starts with alarm..
• The live commands are published unless they were caused by some:excluded-subject – commands without a ditto-originator header are suppressed as well, see absent header behavior for how to keep them.

Two headers which Ditto sets itself tell what caused a signal:

• header:ditto-originator resolves to the first authorization subject of the request that caused the signal.
• header:ditto-origin resolves to the ID of the connection – or of the WebSocket session – through which the causing request entered Ditto. It is absent for signals caused via the HTTP API or by Ditto itself, so that a plain ne comparison would suppress all of those signals, too. To exclude only the signals caused via one connection, supply a value for the absent case: fn-filter=header:ditto-origin|fn:default('none')|fn:filter('ne','some-other-connection-id'). Ditto never publishes a signal to the connection which caused it, so no filter is needed for that; filtering on ditto-origin is only of use to exclude signals caused via other connections.

Combining conditions with AND

An fn-filter tests one value with one fn:filter. To require several conditions, repeat the fn-filter parameter – every expression must match for the topic to be published:

fn-filter=topic:action|fn:filter('ne','deleted')&fn-filter=header:ditto-originator|fn:filter('ne','some:subject')

Keep in mind that each expression suppresses the topic when its placeholder does not resolve for a signal (see absent header behavior).

An RQL filter can be combined with fn-filter parameters on one topic, again with AND semantics – the RQL expression and every pipeline must match:

filter=gt(attributes/counter,42)&fn-filter=topic:action|fn:filter('ne','deleted')&fn-filter=header:ditto-originator|fn:filter('ne','some:subject')

filter may be given at most once per topic – combine several RQL conditions into a single expression with and(...).

Combining conditions with OR

To publish when one or another header is present, let fn:default(<placeholder>) fall back to the second header when the first is absent:

fn-filter=header:first|fn:default(header:second)|fn:filter('exists','true')

The same fallback works with any rqlFunction, but keep in mind that it is a coalesce: the fn:filter tests the first present value only – header:second is not looked at when header:first is present.

For an OR of arbitrary conditions, list the topic several times in the target’s topics array, each entry with its own filters. A signal is published – once – as soon as it matches any one entry:

{
  "address": "<target-address>",
  "topics": [
    "_/_/things/twin/events?fn-filter=header:ditto-originator|fn:filter('eq','some:subject')",
    "_/_/things/twin/events?fn-filter=resource:path|fn:filter('like','/features/*')"
  ],
  "authorizationContext": ["ditto:outbound-auth-subject"]
}

This target publishes the events which were caused by some:subject or whose resource path starts with /features/.

If such entries define different extraFields and several of them match, the signal is enriched with the extraFields of one matching entry only – use the same extraFields on all entries of an OR.

Absent header behavior

Since the pipeline is evaluated per signal, its leading placeholder may not resolve for a given signal. An absent header is the common case – ditto-originator is absent for signals with no authenticated causing subject, ditto-origin for all signals which were not caused via a connection or a WebSocket session – but the same goes for any other placeholder: topic:subject only resolves for messages, feature:id only for signals related to a feature, and thing-json:... only for thing events.

A placeholder which does not resolve always suppresses the topic, regardless of the rqlFunction: there is no value to filter, so the fn:filter never matches – also for ne. The same holds for a placeholder used as <comparedValue> which does not resolve.

To publish in that case, opt in explicitly by supplying a value for it with an fn:default(...) stage before the fn:filter:

fn-filter=header:ditto-originator|fn:default('none')|fn:filter('ne','some:subject')
fn-filter=header:ditto-originator|fn:default('none')|fn:filter('eq','none')

The first publishes signals which carry another originator or none at all, the second only signals without an originator (pick a default value which cannot occur as a real value).

Restrictions

All of the following is checked at connection creation/update time and rejected with HTTP status 400: an invalid fn-filter, or a placeholder pipeline placed in filter, with error connectivity:connection.configuration.invalid; a malformed RQL filter with error rql.expression.invalid; a repeated filter, namespaces or extraFields parameter with error connectivity:topic.invalid. The message names the offending expression or topic; for a malformed RQL filter it is the message of the RQL parser.

• filter only accepts an RQL expression – a placeholder pipeline is rejected with an error pointing to fn-filter.
• fn-filter must start with a placeholder (header:..., topic:..., thing-json:..., …) and every further stage must be an fn: function call. An expression starting with fn:, an RQL expression, or a leading placeholder without a name (e.g. header:) or with whitespace next to its colon (e.g. header: x) is rejected, as is everything else the placeholder grammar does not accept: an unknown placeholder or function, malformed function parameters, or an empty fn-filter=.
• An fn-filter contains exactly one fn:filter(...) stage, as its last stage: a bare placeholder (fn-filter=header:ditto-originator; write header:ditto-originator|fn:filter('exists','true') instead), a pipeline ending with a value-producing stage (e.g. fn:upper()) or with an fn:default(...) (which would discard the filter’s decision), and a second fn:filter stage (repeat the fn-filter parameter instead) are rejected. So is an fn:delete() stage at any position, because the topic would never publish.
• The fn:filter stage must have the form fn:filter('<rqlFunction>',<comparedValue>): the 3-parameter form and fn:filter(<value>,'exists'), which pass the value to filter as a parameter, are rejected, as is a placeholder used as rqlFunction. The rqlFunction name must be one of the case-sensitive eq, ne, like, exists RQL functions (so 'NE' or 'neq' are rejected). The constant compared value 'exists' cannot be used (fn:filter would take it for its fn:filter(<value>,'exists') form). With the rqlFunction 'exists' the compared value must be the constant 'true' (in upper or lower case): a placeholder and any other constant are rejected – fn:filter('exists','false') could only match an empty value, never an absent one, because the filtered value is always a resolved one; see absent header behavior for how to publish on an absent value.
• An fn-filter may contain at most 10 fn: stages, the final fn:filter included; the number of fn-filter parameters is not limited.
• filter, namespaces and extraFields may be given at most once per topic; repeating one of them is rejected as an invalid topic. Only fn-filter is repeatable.
• The thing-json placeholder can only be used as the leading placeholder (thing-json:attributes/x|fn:filter('eq','y')), not as <comparedValue> or inside fn:default(...), because a function parameter’s placeholder prefix must not contain a dash.

Further behavior to be aware of, which is not validated:

• The placeholders of an fn-filter never see fields added via extraFields enrichment – they only ever see what the signal itself carries. Unlike RQL, an fn-filter cannot filter on enriched, unchanged data. The thing-json placeholder sees only the data carried by the event itself; for live commands and messages it never resolves.
• A placeholder used as <comparedValue> of eq, ne or like must never resolve to the string exists: fn:filter would take it for its fn:filter(<value>,'exists') form and match every value.
• An fn-filter should only reference headers that are stable for the signal’s lifetime (such as ditto-originator or ditto-origin): internal bookkeeping headers such as requested-acks are mutated while the signal is processed and are not reliable filter inputs.
• As with RQL, neither filter nor fn-filter can be set on _/_/policies/announcements or _/_/connections/announcements (see the table above); both are silently ignored if present.
• A signal suppressed by a filter is not logged as suppressed, the same as with an RQL filter: the connection logs merely lack the filtered entry for that target. Only when evaluating an fn-filter fails (rather than simply not matching) is a failure entry recorded there; the signal is then treated as not matching that topic.

Target topics and enrichment

You can add extra fields to outgoing messages with the extraFields parameter. See signal enrichment for details.

Not all topics support enrichment:

Example:

{
  "address": "<target-address>",
  "topics": [
    "_/_/things/twin/events?extraFields=attributes/placement",
    "_/_/things/live/messages?extraFields=features/ConnectionStatus"
  ],
  "authorizationContext": ["ditto:outbound-auth-subject"]
}

Target issued acknowledgement label

A target can automatically issue an acknowledgement once the channel confirms successful delivery. The label must be prefixed by the connection ID or {{connection:id}}:

{
  "address": "<target>",
  "topics": ["_/_/things/twin/events"],
  "authorizationContext": ["ditto:outbound-auth-subject"],
  "issuedAcknowledgementLabel": "{{connection:id}}:my-custom-ack"
}

Target header mapping

You can apply an optional header mapping to outgoing messages:

{
  "address": "<target>",
  "topics": ["_/_/things/twin/events"],
  "authorizationContext": ["ditto:outbound-auth-subject"],
  "headerMapping": {
    "message-id": "{{ header:correlation-id }}",
    "content-type": "{{ header:content-type }}",
    "subject": "{{ topic:subject }}",
    "reply-to": "all-replies"
  }
}

Authorization

Ditto initiates connections as a client, so no client authorization is needed from the external system. However, to access Ditto resources, each connection must specify an authorizationContext with self-assigned authorization subjects. These subjects must be granted access through Policies.

• A target can only send data for things to which it has READ rights
• A source can only receive data for things to which it has WRITE rights

Placeholders

Connection configurations support placeholders with the syntax {{ placeholder }}. See the placeholders concept for full details.

Placeholder for source authorization subjects

You can use header placeholders in source authorization subjects to apply per-device permissions:

{
  "id": "auth-subject-placeholder-example",
  "sources": [{
    "addresses": ["telemetry"],
    "authorizationContext": ["device:{{ header:device_id }}"]
  }]
}

Placeholder for target addresses

You can use thing placeholders in target addresses to route messages by namespace or device:

All connection placeholders are also available. If any placeholder fails to resolve, the message is dropped.

{
  "id": "target-placeholder-example",
  "targets": [{
    "addresses": ["live/{{ thing:namespace }}"],
    "authorizationContext": ["ditto:auth-subject"],
    "topics": ["_/_/things/live/events", "_/_/things/live/commands"]
  }]
}

Specific configuration

Some connection types require protocol-specific settings in the specificConfig field. See each protocol binding page for details.

Payload mapping

You can transform message payloads between external formats and Ditto Protocol using payload mapping.

SSH tunneling

Ditto supports tunneling connections through SSH. See SSH tunneling for setup instructions.

Further reading

• Managing connections – create, modify, and monitor connections
• Payload mapping – transform message payloads
• Header mapping – map external headers
• TLS certificates – secure connections with TLS
• Acknowledgements – configure delivery guarantees