Connections

Connection model

You can integrate your Ditto instance with external messaging services such as Eclipse Hono, a RabbitMQ broker or an Apache Kafka broker via custom “connections”.

Additionally, you may invoke foreign HTTP endpoints by using the HTTP connection type.

A connection represents a communication channel for the exchange of messages between any service and Ditto. It requires a transport protocol, which is used to transmit Ditto Protocol messages. Ditto supports one-way and two-way communication over connections. This enables consumer/producer scenarios as well as fully-fledged command and response use cases. Nevertheless, those options might be limited by the transport protocol or the other endpoint’s capabilities.

All connections are configured and supervised via Ditto’s Connectivity service. The following model defines the connection itself:

Connection types

The top design priority of this model is to be as generic as possible, while still allowing protocol specific customizations and tweaks. This enables the implementations of different customizable connection types, and support for custom payload formats. Currently the following connection types are supported:

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

The sources and targets address formats depends on the connectionType and has therefore connectionType specific limitations. Those are documented with the corresponding protocol bindings.

Sources

Sources are used to connect to message brokers / external systems in order to consume messages from them.

Source messages can be of the following type:

• commands
• messages
• live commands/responses/events

Sources contain:

• several addresses (depending on the connection type those are interpreted differently, e.g. as queues, topics, etc.),
• a consumer count defining how many consumers should be attached to each source address,
• an authorization context (see authorization) specifying which authorization subject is used to authorize messages from the source,
• enforcement information that allows filtering the messages that are consumed in this source,
• header mapping for mapping headers of source messages to internal headers, and
• a reply-target to configure publication of any responses of incoming commands.

Source enforcement

Messages received from external systems are mapped to Ditto internal format, either by applying some custom mapping or the default mapping for Ditto Protocol messages.

During this mapping the digital twin of the device is determined i.e. which Thing is accessed or modified as a result of the message. By default no sanity check is done if this target Thing corresponds to the device that originally sent the message. In some use cases this might be valid, but in other scenarios you might want to enforce that a device only sends data to its digital twin. Note that this could also be achieved by assigning a specific policy to each device and use placeholders in the authorization subject, but this can get cumbersome to maintain for a large number of devices.

With an enforcement you can use a single policy for all devices and still make sure that a device only modifies its associated digital twin. Enforcement is only feasible if the message contains the verified identity of the sending device (e.g. in a message header). This verification has to be done by the external system e.g. by properly authenticating the devices and providing the identity in the messages sent to Ditto.

The enforcement configuration consists of two fields:

• input: Defines where device identity is extracted.
• filters: Defines the filters that are matched against the input. At least one filter must match the input value, otherwise the message is rejected.

The following placeholders are available for the input field:

The following placeholders are available for the filters field:

Assuming a device sensor:temperature1 pushes its telemetry data to Ditto which is stored in a Thing sensor:temperature1. The device identity is provided in a header field device_id. To enforce that the device can only send data to the Thing sensor:temperature1 the following enforcement configuration can be used:

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

Note: This example assumes that there is a valid user named ditto:inbound-auth-subject in Ditto. If you want to use a user for the basic auth (from the HTTP API) use the prefix nginx:, e.g. nginx:ditto. See Basic Authentication for more information.

Source header mapping

For incoming messages, an optional header mapping may be applied. Mapped headers are added to the headers of the Ditto protocol message obtained by payload mapping. The default Ditto payload mapper does not retain any external header; in this case all Ditto protocol headers come from the header mapping.

The JSON for a source with header mapping could look like this:

{
  "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 the responses of incoming commands. For a reply target, the address and header mapping are defined in itself, whereas its payload mapping is inherited from the parent source, because a payload mapping definition specifies the transformation for both incoming and outgoing messages.

For example, to publish responses at the target address equal to the reply-to header of incoming commands, define source header mapping and reply target as follows. If an incoming command does not have the reply-to header, then its response is dropped.

{
  "headerMapping": {
    "reply-to": "{{ header:reply-to }}"
  },
  "replyTarget": {
    "enabled": true,
    "address": "{{ header:reply-to }}"
  }
}

Targets

Targets are used to connect to messages brokers / external systems in order to publish messages to them.

Target messages can be of the following type:

• messages
• events
• live commands/responses/events

Targets contain:

• one address (that is interpreted differently depending on the connection type, e.g. as queue, topic, etc.),
• topics that will be sent to the target,
• an authorization context (see authorization) specifying which authorization subject is used to authorize messages to the target, and
• header mapping to compute external headers from Ditto protocol headers.

Target topics and filtering

For targets it can be configured which types of messages should be published to the target address.

In order to only consume specific events like described in change notifications, the following parameters can additionally be provided when specifying the topics of a target:

The parameters are specified similar to HTTP query parameters, the first one separated with a ? and all following ones with &. You have to URL encode the filter values before using them in a configuration.

For example this way the connection session would register for all events in the namespace org.eclipse.ditto and which would match an attribute “counter” to be greater than 42. Additionally it would subscribe to messages in the namespace org.eclipse.ditto:

{
  "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

When extra fields should be added to outgoing messages on a connection, an extraFields parameter can be added to the topic. This is supported for all topics:

Example:

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

Target header mapping

For outgoing messages, an optional header mapping may be applied. Mapped headers are added to the external headers. The default Ditto payload mapper does not define any external header; in this case all external headers come from the header mapping.

The JSON for a target with header mapping could like this:

{
  "address": "<target>",
  "topics": [
    "_/_/things/twin/events",
    "_/_/things/live/messages?namespaces=org.eclipse.ditto"
  ],
  "authorizationContext": ["ditto:inbound-auth-subject"],
  "headerMapping": {
    "message-id": "{{ header:correlation-id }}",
    "content-type": "{{ header:content-type }}",
    "subject": "{{ topic:subject }}",
    "reply-to": "all-replies"
  }
}

Authorization

A connection is initiated by the connectivity service. This obviates the need for client authorization, because Ditto becomes the client in this case. Nevertheless, to access resources within Ditto, the connection must know on whose behalf it is acting. This is controlled via the configured authorizationContext, which holds a list of self-assigned authorization subjects. Before a connection can access a Ditto resource, one of its authorizationSubjects must be granted the access rights by an authorization mechanism such as ACLs or Policies.

A connection target can only send data for Things to which it has READ rights, as data flows from a Thing to a target. A connection source can only receive data for Things to which it has WRITE rights, as data flows from a source to a Thing.

Specific config

Some connection types require specific configuration which are not supported for other connection types. Those are put into the specificConfig field.

Payload Mapping

For more information on mapping message payloads see the corresponding Payload Mapping Documentation.

Placeholders

The configuration of a connection allows to use placeholders at certain places. This allows more fine grained control over how messages are consumed or where they are published to. The general syntax of a placeholder is {{ placeholder }}. Have a look at the placeholders concept for more details on that.

Placeholder for source authorization subjects

Processing the messages received via a source using the same fixed authorization subject may not be suitable for every scenario. For example, if you want to declare fine-grained write permissions per device, this would not be possible with a fixed global subject. For this use case we have introduced placeholder substitution for authorization subjects of source addresses that are resolved when processing messages from a source. Of course, this requires the sender of the message to provide necessary information about the original issuer of the message.

You can access any header value of the incoming message by using a placeholder like {{ header:name }}.

Example:

Assuming the messages received from the source telemetry contain a device_id header (e.g. sensor-123), you may configure your source’s authorization subject as follows:

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

The placeholder is then replaced by the value from the message headers and the message is forwarded and processed under the subject device:sensor-123. In case the header cannot be resolved or the header contains unexpected characters an exception is thrown which is sent back to the sender as an error message, if a valid reply-to header was provided, otherwise the message is dropped.

Placeholder for target addresses

Another use case for placeholders may be to publish Thing events or live commands and events to a target address containing Thing-specific information e.g. you can distribute Things from different namespaces to different target addresses. You can use the placeholders {{ thing:id }}, {{ thing:namespace }} and {{ thing:name }} in the target address for this purpose. For a Thing with the ID org.eclipse.ditto:device-123 these placeholders would be resolved as follows:

Even more than the ones above, all mentioned connection placeholders may be used in target addresses. However, if any placeholder in the target address fails to resolve, then the message will be dropped.

Example:

Sending live commands and events to a target address that contains the Things’ namespace.

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