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 and RQL expressions:

Filter parameters use HTTP query parameter syntax (? for the first, & for subsequent). URL-encode filter values before using them:

{
  "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"
  ],
  "authorizationContext": ["ditto:outbound-auth-subject"]
}

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