Data Stream Bridge

Forward selected WEDA DataStream values to an MQTT broker used by your application. A Data Stream Bridge reads arriving values, maps them into JSON fields and publishes to the topic you configure.

The walkthrough forwards temperature_a and the Boolean both_abnormal from the Analyzer end-to-end scenario. You can use other existing DataStreams instead. The Analyzer calculates the result; the Bridge delivers it in the format your application needs.

Understand Data Forwarding

The connection runs from WEDA Cloud to your MQTT broker. The broker must be reachable from Cloud, even when it is located at your factory. This forwarding path does not require the edge device to connect directly to that broker.

Three resources have different jobs:

Resource	What you configure
MQTT Broker Config	Reusable broker endpoint, credentials, protocol and TLS trust. Several bridges can use one configuration.
CA certificate	An uploaded public certificate for a private or self-signed broker CA. The broker configuration refers to its `certificateId`.
Bridge	Source DataStream bindings, destination topic, JSON field mapping and assembly behavior.

Before you start:

Prepare your WEDA API base URL, target Org ID and an authorized user access token.
Obtain the broker endpoint, publishing credentials, allowed topic and certificate requirements from the broker administrator.
Identify the source Device IDs and DataStream IDs, and check that the streams have new values. A Bridge can use device-originated streams or Analyzer outputs.
Replace https://weda.example.com/central/weda, route placeholders, example IDs and broker credentials with your deployment's values.

Note: Console and API coverage

If your deployed WEDA Console includes MQTT Broker Configs and DataStream Bridges, select the target Org before using those views. The API steps below cover the complete workflow, including TLS, CA certificates and assembly settings that may not have corresponding Console controls.

Configure an MQTT Broker

Create a reusable connection profile with Create MQTT Broker Config. Choose a name unique within the Org.

This example saves an mqtts:// profile first. Without a custom CA reference, TLS uses the runtime's operating-system trust store. The next section shows how to attach a private CA and test the connection before enabling the Bridge.

POST https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/mqtt-broker-configs
Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

{
  "name": "Factory MQTT",
  "brokerEndpoint": "mqtts://mqtt.example.com:8883",
  "authType": "UsernamePassword",
  "brokerCredentials": {
    "username": "weda-publisher",
    "password": "YOUR_BROKER_PASSWORD"
  },
  "mqttProtocolVersion": "V311",
  "keepAliveInterval": "60s",
  "reconnectInterval": "5s",
  "checkConnection": false
}

Setting	Meaning
`authType`	`UsernamePassword` is the supported authentication type, including for TLS connections.
`mqttProtocolVersion`	`V310`, `V311` or `V500`. The default is `V311`.
`keepAliveInterval` / `reconnectInterval`	Duration strings. When omitted, the stored values can be `null`; runtime defaults are `60s` and `5s`.
`checkConnection`	`true` tests the broker connection synchronously. `false`, the default, saves the configuration without that live test.

Check the result: After 201 Created, retain mqttConfigId. Use Get MQTT Broker Config to check the endpoint, tlsEnabled, hasCustomCaCertificate and hasStoredCredentials. A saved profile with checkConnection: false does not prove that the broker is reachable. Read responses do not return credential secrets.

Configure TLS and CA Certificates

TLS is enabled by the mqtts:// endpoint scheme, while authentication remains UsernamePassword. WEDA requires TLS 1.2 or later.

Broker configuration	Trust behavior
`mqtts://` without `tls.caCertificateId`	Verify the broker against the operating-system trust store.
`mqtts://` with `tls.caCertificateId`	Verify against the uploaded CA reference.
`mqtt://` without `tls`	Use an unencrypted MQTT connection.
`mqtt://` with a CA reference	Reject the configuration with a TLS scheme-mismatch error.

1. Upload a private CA if required. Use Upload MQTT CA Certificate. If the broker uses a certificate trusted by the operating system, skip the upload and test the profile without a tls object.

Upload one public CA certificate in PEM or DER format, at most 40 KB (40,960 bytes). Private keys and certificate-chain bundles are rejected; the filename extension does not determine the format.
Use a certificate name of 1–128 characters, unique within the Org.
Use an identity with access to the Org. The currently available Org role is Admin; use that role for certificate management.

The upload uses multipart/form-data. For example, with a local factory-mqtt-ca.pem file:

curl --request POST "https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/mqtt-ca-certificates" \
  --header "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  --form "name=Factory MQTT CA" \
  --form "certificate=@factory-mqtt-ca.pem"

Check the result: After 201 Created, retain certificateId and inspect expiresAt and isExpired. You can find it again with List MQTT CA Certificates. Certificate content is not returned.

2. Attach the CA and verify the connection. Use Update MQTT Broker Config. Resend the intended non-credential settings, add the returned CA ID and set checkConnection: true:

PUT https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/mqtt-broker-configs/{mqttConfigId}
Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

{
  "name": "Factory MQTT",
  "brokerEndpoint": "mqtts://mqtt.example.com:8883",
  "tls": {
    "caCertificateId": "{certificateId}"
  },
  "mqttProtocolVersion": "V311",
  "keepAliveInterval": "60s",
  "reconnectInterval": "5s",
  "checkConnection": true
}

For operating-system trust, omit tls and keep checkConnection: true. The broker certificate must cover the host in brokerEndpoint; a private CA reference does not bypass hostname or validity checks.

Note: Upload success is not a TLS connection test

An already expired certificate can be stored with isExpired: true. Trust, validity and hostname checks happen when connecting, including with checkConnection: true and when enabling a Bridge. Upload a valid certificate and resolve connection-test errors before continuing.

Create a Bridge and Bind DataStreams

Choose the destination topic and the source values your application needs. In a Console Bridge form, when available, the corresponding fields are the Bridge name, Broker Config, Output Topic, QoS and Device → DataStream → Target JSONPath bindings. Use the API for additional assembly options.

Use List DataStreams to obtain each source ID. In the Analyzer scenario, the illustrative IDs are:

Source stream	Type	Illustrative `dataStreamId`	Output path
`temperature_a`	Number, °C	`d4444444-4444-4444-8444-444444444444`	`$.temperatureA`
`both_abnormal`	Boolean	`e5555555-5555-4555-8555-555555555555`	`$.bothAbnormal`

Replace those IDs with the actual IDs from your device. Both happen to belong to the same SubNode in this example; bindings can also refer to different source devices.

Create the definition with Create Bridge:

POST https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges
Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

{
  "name": "Factory temperature status",
  "mqtt": {
    "mqttBrokerConfigId": "{mqttConfigId}",
    "outputTopic": "factory/line-1/temperature-status",
    "qos": 1
  },
  "bindings": [
    {
      "deviceId": "{deviceId}",
      "dataStreamId": "d4444444-4444-4444-8444-444444444444",
      "targetJsonPath": "$.temperatureA",
      "requiredForAssembly": true
    },
    {
      "deviceId": "{deviceId}",
      "dataStreamId": "e5555555-5555-4555-8555-555555555555",
      "targetJsonPath": "$.bothAbnormal",
      "requiredForAssembly": true
    }
  ],
  "assemblyPolicy": {
    "queueLimit": 100,
    "timeoutSeconds": 5,
    "onTimeout": "DropPartial"
  }
}

Broker reference: Requests use mqtt.mqttBrokerConfigId. A Bridge read response exposes that reference as mqtt.mqttConfigId.
Source selection: deviceId and dataStreamId identify the source. sourceFieldName defaults to valueObject; omit dataType to use the source value type.
Output mapping: targetJsonPath selects the JSON field. Use distinct paths that do not overwrite or contain one another, such as $.temperatureA and $.bothAbnormal.
Delivery options: qos accepts 0, 1 or 2, with default 0. The example selects 1.

Check the result: After 201 Created, retain bridgeId. Read it with Get Bridge, and verify the source IDs, output paths, topic and policy. A new Bridge is disabled; creation does not start forwarding.

Configure Payload and Assembly Behavior

With the two required values available, the example publishes a JSON message such as:

{
  "temperatureA": 81.25,
  "bothAbnormal": true
}

The assembly policy determines when values form a message:

Setting	Behavior	Default
`bindings[].requiredForAssembly`	Wait for a value from this binding before producing a complete message.	`false`
`assemblyPolicy.queueLimit`	Limit buffered input and pending required values. Capacity can be reached before a complete message is available.	`100`
`assemblyPolicy.timeoutSeconds`	Wait for an incomplete required-field set. Later arrivals do not restart the current wait. Valid range: 1–300 seconds.	`5`
`assemblyPolicy.onTimeout`	`PublishPartial` sends the available fields; `DropPartial` discards the incomplete set.	`PublishPartial`

Required values are paired in arrival order, one queued value per required field. They do not need to come from the same device, and assembly does not guarantee matching sample timestamps.
An optional field is included when a new value is waiting for it, then consumed. Its last value is not automatically repeated in later messages. With no required fields, an arriving optional value can publish immediately.
This example marks both fields required and uses DropPartial, so a missing temperature or indicator does not produce a complete-looking status message. A legitimate Boolean false is a value; it is different from missing data.
Omit the whole assemblyPolicy on create or full replacement to use defaults. If you provide it, include onTimeout. A partial Bridge update can change just one policy field.

For example, use Patch Bridge to increase the wait to ten seconds:

PATCH https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/{bridgeId}
Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

{
  "assemblyPolicy": {
    "timeoutSeconds": 10
  }
}

Check the result: Read the definition back and verify the effective policy. Subscribe to the configured topic with your MQTT client, let new source values arrive, and check the actual fields and types in the received JSON. A Console payload preview, when available, is not proof of a broker delivery.

Caution: Buffer capacity can drop values

queueLimit limits buffering; publishing is driven by required-field readiness and timeout. Values can be dropped when capacity is reached. Choose reporting rates, a timeout and a queue limit that fit your sources, and investigate a required stream that stops reporting.

Bridge assembly formats values for delivery. Keep the temperature conversion and the two-temperature condition in the Analyzers; configuring fields in a Bridge does not perform that calculation.

Enable and Monitor a Bridge

1. Enable forwarding. Use Enable or Disable Bridge:

POST https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/{bridgeId}:enable
Authorization: Bearer YOUR_ACCESS_TOKEN

Enabling performs a live check of the linked broker configuration and credentials. An unreachable broker or failed TLS check can reject the request. 204 No Content has no response body; read the runtime state next.

2. Read the current state. Use Get Bridge:

GET https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/{bridgeId}
Authorization: Bearer YOUR_ACCESS_TOKEN

Detail field	What to check
`isEnabled`	Whether forwarding is requested for this Bridge.
`runtimeState.status`	`Stopped`, `Starting`, `Running`, `Faulted` or `Stopping`.
`runtimeState.lastPublishTime`	Time of the last published payload; it can be `null` before the first publish.
`runtimeState.queueDepth`	Required values still waiting in the assembly queues.
`runtimeState.fault`	Error details when faulted: `code`, `message`, `willRetry` and `maxRetryAttempts`.

The List Bridges response uses the shorter runtimeStatus field. The detail response uses runtimeState.status.

Check the result: Confirm isEnabled: true and a Running runtime, then verify a new message at the destination broker. Inspect lastPublishTime as values arrive. Running alone does not confirm that your selected source streams have produced data or that your application consumed it.

Note: An enabled Bridge can briefly report Stopped

Updating a broker configuration or rotating its credentials automatically restarts the enabled bridges that use it. During reconnection, runtimeState.status can briefly be Stopped while isEnabled remains true. Refresh the state and distinguish this transition from a Bridge that you disabled.

Update, Delete and Troubleshoot

Use the operation that matches what you want to change:

Task	API and behavior
Replace a Bridge definition	Update Bridge, `PUT .../data-bridges/{bridgeId}`. Send the complete intended definition.
Change selected Bridge fields	Patch Bridge. Including `bindings` replaces the entire bindings array.
Change broker endpoint, protocol or CA trust	Update MQTT Broker Config. This replaces the non-credential settings.
Rotate broker credentials	Update MQTT Broker Credentials, `PATCH .../mqtt-broker-configs/{mqttConfigId}/credentials`.

Caution: Preserve the CA reference on broker updates

An MQTT Broker Config PUT clears the stored CA reference when you omit tls or set its CA ID to null. An mqtts:// connection then uses operating-system trust instead. Read the current profile first and resend its tls.caCertificateId on every replacement that must retain the custom CA.

The update also restarts enabled bridges using that profile, even if the submitted values are unchanged. Recheck runtime state and a received message after the update.

For credential rotation, send the supported authType and the new credentials through the dedicated endpoint:

PATCH https://weda.example.com/central/weda/api/v1/orgs/{orgId}/data-bridges/mqtt-broker-configs/{mqttConfigId}/credentials
Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

{
  "authType": "UsernamePassword",
  "brokerCredentials": {
    "username": "weda-publisher",
    "password": "YOUR_NEW_BROKER_PASSWORD"
  },
  "checkConnection": true
}

To remove the resources, work in dependency order:

If the Bridge is enabled, call POST .../data-bridges/{bridgeId}:disable and read its state back. Disabling stops forwarding and retains the definition.
Delete the disabled Bridge. An enabled Bridge cannot be deleted.
Delete the broker configuration only after its referencing bridges have been removed or repointed. A disabled Bridge still references its configuration.
Delete the CA certificate after broker configurations no longer refer to it. Any broker reference in the tenant blocks deletion. Use the certificate list's isInUse indicator to check its current use.

After a 204 No Content deletion, list the corresponding resources to confirm removal. A blocked dependency deletion returns 400 in the current public contract. A duplicate broker or Bridge name returns 409.

When forwarding fails, start with the runtime fault and the actual broker result:

Symptom or fault	What to check
`BROKER_CONNECTION_FAILED`	WEDA Cloud can reach the broker host/port; credentials and broker-side permissions are correct; an `mqtts://` broker supports TLS 1.2 or later.
`CREDENTIALS_NOT_FOUND`	Store valid broker credentials through the credentials endpoint.
`TLS_INVALID_CA_CERT`	The referenced CA exists and can validate the broker certificate chain.
`TLS_EXPIRED_CA_CERT`	Check the validity periods of the broker certificate and its chain.
`TLS_HOSTNAME_MISMATCH`	The host in `brokerEndpoint` is covered by the broker certificate.
Running, but no received message	Confirm new source values, correct stream IDs, binding paths, required-field arrival, timeout policy and the subscriber's topic/permissions.
Queue grows or messages lack expected fields	Check mismatched reporting rates, missing required sources and optional-field consumption. Increasing capacity alone does not restore a missing source.

Read fault.message alongside fault.code. If willRetry is true, monitor reconnection; when automatic retries have stopped, correct the configuration and use a disable/enable cycle to retry. See Get Bridge for the documented fault codes.

Understand Data Forwarding​

Configure an MQTT Broker​

Configure TLS and CA Certificates​

Create a Bridge and Bind DataStreams​

Configure Payload and Assembly Behavior​

Enable and Monitor a Bridge​

Update, Delete and Troubleshoot​

Is this helpful?