Azure Event Hubs Bridge
Configure this plugin in the Platform UI. Changes apply without a broker restart and are synchronized across cluster nodes. See Bridges.
The Azure Event Hubs Bridge plugin forwards MQTT messages from the Mosquitto broker to one or more Azure Event Hubs.
The plugin supports:
- Topic routing with 1:1, 1:n, m:1, and m:n mappings between MQTT topics and event hubs. See Topic mappings.
- Authentication with a Shared Access Signature (SAS) connection string or Role-Based Access Control (RBAC). See Event hubs.
- Partition ordering with
partitionIdor apartitionKeytaken from a JSON payload. See Producers. {NOW}and{NOW_ISO}placeholders inmessageIdandcorrelationId. See Producers.- Immediate (
direct) or buffered forwarding. See Producers. - Retry settings. See Producers.
- Optional restructuring of the event body, including MQTT metadata. See Schema mappings.
Create the Azure Event Hubs namespace, hubs, and credentials in Azure before configuring the plugin. The plugin does not create Azure resources.
Plugin activation
A license that includes the Azure Event Hubs Bridge is required. Add the plugin to mosquitto.conf:
plugin /usr/lib/cedalo_azure_eventhubs.so
persistence_location /mosquitto/data
This path applies to the Mosquitto broker container image. For other installations, set plugin to the location of cedalo_azure_eventhubs.so.
Configuration
The configuration is a JSON object. eventhubs and producers are required. topicMappings selects which MQTT messages are forwarded. schemaMappings is optional and reshapes the event body.
eventhubs: Azure Event Hubs namespaces. See Event hubs.producers: how events are sent to a hub. See Producers.topicMappings: which MQTT topics are forwarded, and to which producer. See Topic mappings.schemaMappings: optional payload and metadata mapping. See Schema mappings.
The JSON schema lists every configuration property. Request and response payloads are specified in the MQTT API.
Configuration file format
Event hubs
Each entry names an Event Hubs namespace and selects one authentication method.
For SAS, auth.sas.key is the Event Hubs connection string. The string already contains the endpoint, so fullyQualifiedNamespace is not used.
For RBAC, tenantId, clientId, and clientSecret are required. fullyQualifiedNamespace is optional. If it is omitted, the plugin uses <name>.servicebus.windows.net, where name is the namespace entry's name.
RBAC example:
{
"name": "EventHubsResource",
"auth": {
"rbac": {
"tenantId": "12345677-6171-47aa-b47b-1234567d89de",
"clientId": "abcdef12-aa90-4190-9cfc-123456789abc",
"clientSecret": "aVerySecretClientSecret",
"fullyQualifiedNamespace": "resource_namespace.servicebus.windows.net"
}
}
}
SAS example:
{
"name": "EventHubsResource",
"auth": {
"sas": {
"key": "Endpoint=sb://resource_namespace.servicebus.windows.net/;SharedAccessKeyName=sender;SharedAccessKey=secret;EntityPath=hub1"
}
}
}
Producers
A producer sends events to one hub inside a namespace. eventhub is the name of an entry in eventhubs. Authentication comes from that namespace. Set auth on the producer only to override it. A producer override is merged into the namespace credentials, so an RBAC override can set fullyQualifiedNamespace alone.
type is direct or buffered. A direct producer sends each event immediately. A buffered producer collects events and sends them together. If type is omitted, the producer is direct.
Older configurations are still accepted. type value batch is treated as direct. A batch object is read as direct, and maxOpenBatches is read as maxActiveEvents. A buffer object is read as buffered.
Producer properties:
name: unique name used by topic mappings.hub: event hub name inside the namespace.type:directorbuffered. Default:direct.eventhub:nameof aneventhubsentry.auth: optional credentials that override the namespace. Usesasorrbac, with the same fields as a namespace.options: properties applied to every event.messageId: event message id.{NOW}is replaced with the current Unix time in milliseconds.{NOW_ISO}is replaced with an ISO 8601 timestamp. A backslash before a placeholder keeps that text unchanged.correlationId: event correlation id. The same placeholders asmessageIdare supported.partitionId: partition id, such as"0"or"1". Events with the same id are ordered. Do not setpartitionKeyas well.partitionKey: partition key for ordered processing. Do not setpartitionIdas well.field: JSONPath evaluated against a JSON payload, for example$.device.id.default: value used whenfieldis set but does not match. Ignored when the payload is not JSON.
retryOptions: copied onto the Azure Event Hubs client.maxRetries: retry attempts after the first send fails. Azure default: 3.retryDelayInMs: delay in milliseconds before the next attempt. Azure default: 30000. With an exponential delay this is the base interval.maxRetryDelayInMs: upper bound for an exponential delay.timeoutInMs: time in milliseconds before the current attempt is treated as timed out. Azure raises values below 60000 to 60000.mode:0selects an exponential delay. Any other value, including a missingmode, selects a fixed delay. The configuration schema only allows the stringsfixandexponential, soupdateConfigrejects0.
direct: options for a direct producer.maxSizeInBytes: maximum batch size in bytes. If omitted, Azure Event Hubs selects the size.maxActiveEvents: maximum number of batches sending or retrying at the same time. Default: 10.
buffered: options for a buffered producer. These values are passed to the Azure buffered producer.maxWaitTimeInMs: time in milliseconds to wait for another event. If none arrives, the buffered events are sent. Azure default: 1000.maxEventBufferLengthPerPartition: maximum number of events buffered for one partition before they are sent. Azure default: 1500.
Azure Event Hubs does not read delayInMs, maxDelayInMs, or the mode strings fix and exponential. Those names are still in the configuration schema. exponential does not select an exponential delay.
Set maxEventBufferLengthPerPartition. A getSchema response can still list maxEventBufferLengthPerPartitions. That spelling is not applied.
The following producer uses the namespace above, overrides fullyQualifiedNamespace, and buffers events for up to one second:
{
"name": "Producer1",
"type": "buffered",
"hub": "hub1",
"eventhub": "EventHubsResource",
"options": {
"messageId": "Event sent at {NOW_ISO}"
},
"auth": {
"rbac": {
"fullyQualifiedNamespace": "another_resource_namespace.servicebus.windows.net"
}
},
"retryOptions": {
"maxRetries": 100,
"retryDelayInMs": 1000,
"timeoutInMs": 60000
},
"buffered": {
"maxWaitTimeInMs": 1000
}
}
Topic mappings
A topic mapping forwards messages that match mqttTopics to the producer named by target. MQTT wildcards are supported: + matches one level and # matches the remaining levels.
Several mappings can express:
1:1: one topic filter to one producer.1:n: one topic filter to several producers, using one mapping per producer.m:1: several topic filters to one producer.m:n: several topic filters to several producers.
m:n example:
[
{
"mqttTopics": ["topic/test1", "topic/test2"],
"target": "Producer1"
},
{
"mqttTopics": ["topic/test1", "topic/test2"],
"target": "Producer2"
}
]
If the same topic filter is mapped to the same producer more than once, the last mapping is used.
schemaMapping selects how the event body is built. Use the name of an entry in schemaMappings. If schemaMapping is omitted, the original MQTT payload is forwarded unchanged.
[
{
"mqttTopics": ["topic/test1", "topic/test2"],
"target": "Producer1",
"schemaMapping": "Mapping1"
},
{
"mqttTopics": ["topic/+/prod/#"],
"target": "Producer2",
"schemaMapping": "Mapping1"
},
{
"mqttTopics": ["topic/test1", "topic/test2"],
"target": "Producer2",
"schemaMapping": "Mapping2"
}
]
The bridge also accepts an inline mapping object with name and mapping. The configuration schema types schemaMapping as a string, so a named entry is the form updateConfig validates.
[
{
"mqttTopics": ["topic/test1", "topic/test2"],
"target": "Producer1",
"schemaMapping": {
"name": "Mapping1",
"mapping": [
{ "source": "[topic]", "target": "[metadata][topic]" },
{ "source": "[payload]", "target": "[payload]" }
]
}
}
]
Schema mappings
With a schema mapping, the event body is a JSON object built from the MQTT message. Without one, the body is the original payload string.
schemaMappings is an optional array. Each entry needs a unique name and a mapping array. name is the value used by topicMappings[].schemaMapping.
Each item in mapping is an extractor. source selects a value. target is the JSON path where that value is written. Every target must use path syntax, for example [reading] or [metadata][topic].
Path segments inside [] may contain letters, digits, _, and -.
A source path starts with one of the metadata names below. Further segments read properties of a JSON payload, for example [payload][temperature]. A source that is not one of these paths is stored as a constant. Set options.isConst to true when the constant itself contains square brackets, so it is not read as a path.
Supported source roots:
hostname: broker hostname. Requires theHOSTNAMEenvironment variable. Otherwise the value is<Unknown>(string).topic: MQTT topic (string).payload: MQTT payload. JSON payloads are parsed and can be unpacked further. Any other payload is kept as a string or number.qos: MQTT QoS,0,1, or2(integer).retain: MQTT retain flag (boolean).timestamp: Unix time in seconds when the message arrived at the broker (integer).timestampMs: Unix time in milliseconds when the message arrived at the broker (integer).datetime: arrival time asyyyy-MM-ddTHH:mm:ss, without fractional seconds and without aZsuffix (string).uuid: id generated by the bridge for this event. It is not a property of the original MQTT message (string).client_id: client id of the publishing client (string).
Example:
[
{
"name": "M1",
"mapping": [
{ "source": "[datetime]", "target": "[metadata][dt]" },
{ "source": "[client_id]", "target": "[metadata][client][id]" },
{ "source": "[retain]", "target": "[metadata][message][ret_flag]" },
{ "source": "[qos]", "target": "[metadata][message][qos]" },
{ "source": "[topic]", "target": "[metadata][message][topic]" },
{ "source": "[payload]", "target": "[payload]" }
]
},
{
"name": "M2",
"mapping": [
{ "source": "[topic]", "target": "[metadata][topic]" },
{ "source": "[payload][temperature]", "target": "[reading]" }
]
}
]
M1 copies metadata and the payload. The event body is:
{
"metadata": {
"dt": "1970-01-01T00:00:00",
"message": {
"ret_flag": false,
"qos": 0,
"topic": "topic/test1"
},
"client": {
"id": "device-1"
}
},
"payload": {
"temperature": 21.5
}
}
M2 copies the topic and the payload property temperature. For a payload {"temperature": 21.5} the event body is:
{
"metadata": {
"topic": "topic/test1"
},
"reading": 21.5
}
A constant is written by putting the value in source and the destination path in target:
[
{
"name": "M3",
"mapping": [
{ "source": "test data", "target": "[label]" },
{ "source": "[payload]", "target": "[payload]" }
]
}
]
This writes the string test data to label. If the constant contains square brackets, set isConst so the value is not treated as a path:
[
{
"name": "M3",
"mapping": [
{ "source": "[test_data]", "target": "[label]", "options": { "isConst": true } },
{ "source": "[payload]", "target": "[payload]" }
]
}
]
This writes the literal string [test_data] to label.
Extractor options
options transforms source before it is written to target.
{
"name": "mapping1",
"mapping": [
{
"source": "[payload][property1]",
"target": "[my_property]",
"options": {
"replaceNullWith": "<null>"
}
}
]
}
Here a JSON null in property1 is written as the string "<null>".
Supported options:
isConst(path:$.schemaMappings[].mapping[].options.isConst): whentrue,sourceis written as a constant. Required when the constant uses[...]path syntax. Omitted for constants that do not look like a path (boolean).replace(path:$.schemaMappings[].mapping[].options.replace): two strings. Every occurrence of the first is replaced with the second. Applies when the value is a string (array of two strings).truncate(path:$.schemaMappings[].mapping[].options.truncate): maximum number of characters kept from a string (integer).replaceNullWith(path:$.schemaMappings[].mapping[].options.replaceNullWith): value written whensourceis JSONnull.replaceUndefinedWith(path:$.schemaMappings[].mapping[].options.replaceUndefinedWith): value written whensourceis missing.nullValue(path:$.schemaMappings[].mapping[].options.nullValue): ifsourceequals this value, the written value is JSONnull.defaultValue(path:$.schemaMappings[].mapping[].options.defaultValue): ifsourceequals this value, the property is omitted from the event body.format(path:$.schemaMappings[].mapping[].options.format): date-fns parse pattern used when casting todatetime. See the date-fns format tokens. If omitted, ISO 8601 strings and Unix timestamps in milliseconds are parsed withDate(string).fromSeconds(path:$.schemaMappings[].mapping[].options.fromSeconds): when casting a number todatetimeandformatis omitted, treat the number as Unix time in seconds (boolean).dbFormat(path:$.schemaMappings[].mapping[].options.dbFormat): date-fns pattern used to format adatetimevalue before it is written (string).replaceDelimiter(path:$.schemaMappings[].mapping[].options.replaceDelimiter): whendbFormatis omitted, adatetimeis written asyyyy-MM-ddTHH:mm:ss. This option replacesTwith the given string, for example a space (string).splitBy(path:$.schemaMappings[].mapping[].options.splitBy): split a string into an array on this delimiter. Can be combined withjoinBy,fetchByIndex, orfetchByRange(string).joinBy(path:$.schemaMappings[].mapping[].options.joinBy): join an array into a string, inserting this delimiter between elements (string).fetchByIndex(path:$.schemaMappings[].mapping[].options.fetchByIndex): zero-based index of one array element to keep (integer).fetchByRange(path:$.schemaMappings[].mapping[].options.fetchByRange): two integers, start inclusive and end exclusive. Can be combined withsplitByorjoinBy. IfsplitByis set andjoinByis not, the slice is joined again with thesplitBydelimiter (array of two integers).
Options that apply only when type is datetime:
formatfromSecondsdbFormatreplaceDelimiter
{
"name": "mapping1",
"mapping": [
{
"source": "[payload][property1]",
"target": "[my_property]",
"type": "datetime",
"options": {
"replaceNullWith": "<null>",
"format": "yyyy-MM-dd HH:mm:ss",
"dbFormat": "dd-MMM-yy HH:mm:ss"
}
}
]
}
Options that apply when the value is a string:
replacetruncatesplitBy
{
"name": "mapping1",
"mapping": [
{
"source": "[payload][property1]",
"target": "[my_property]",
"options": {
"replace": ["hello world", "hello Azure"]
}
}
]
}
Options that apply to arrays, including arrays produced by splitBy:
joinByfetchByIndexfetchByRange
{
"name": "mapping1",
"mapping": [
{
"source": "[payload][property1]",
"target": "[my_property]",
"options": {
"splitBy": "/",
"fetchByRange": [0, 3]
}
}
]
}
Options that apply to any value:
replaceNullWithreplaceUndefinedWithnullValuedefaultValueisConst
Data type casting
Set type on an extractor to cast source before it is written. Cast only values that need it.
{
"name": "mapping1",
"mapping": [
{
"source": "[payload][temperature_str]",
"target": "[temperature]",
"type": "number"
}
]
}
Supported types:
datetime: parse a string or number as a date. Seeformat,fromSeconds,dbFormat, andreplaceDelimiterunder Extractor options. An ISO 8601 string does not need a cast.number: parse a string as a number. Both.and,are accepted as decimal separators, for example1.5and1,5. If the value is not a number, the original value is kept.integer: parse a string as an integer and drop the fractional part. If the value is not a number, the original value is kept.uinteger: same asinteger, then take the absolute value.boolean: the stringfalse(any letter case) and the number0becomefalse. An empty string becomesfalse. Any other present value becomestrue. JSONnulland a missing property are not converted tofalse.string: encode a number, boolean, or object as a string. Strings are left unchanged.
Configuration example
{
"eventhubs": [
{
"name": "EvHub1",
"auth": {
"sas": {
"key": "Endpoint=sb://example.servicebus.windows.net/;SharedAccessKeyName=sender;SharedAccessKey=secret;EntityPath=hub1"
}
}
}
],
"producers": [
{
"name": "P1",
"hub": "hub1",
"eventhub": "EvHub1",
"options": {
"messageId": "{NOW_ISO}"
}
}
],
"topicMappings": [
{
"mqttTopics": ["test/topic"],
"target": "P1",
"schemaMapping": "M1"
}
],
"schemaMappings": [
{
"name": "M1",
"mapping": [
{ "source": "[datetime]", "target": "[metadata][dt]" },
{ "source": "[topic]", "target": "[metadata][topic]" },
{ "source": "[payload]", "target": "[payload]" }
]
}
]
}
P1 uses the default producer type, direct. Messages published to test/topic are sent to hub hub1 after mapping M1 is applied.
Control topic
Publish commands to $CONTROL/cedalo/azure-eventhubs-bridge/v1. Responses are published to $CONTROL/cedalo/azure-eventhubs-bridge/v1/response.
Commands:
getStatus: current bridge status.getSchema: configuration schema, including examples.getConfig: current configuration.updateConfig: replace the configuration. The new object is passed inconfigChange.testProducer: open a connection with the producer settings inparamsand close it.paramsrequiresname,hub, andeventhub.eventhubis a namespace object withnameandauth, not the hub name.
Request shape:
{
"commands": [
{
"command": "<command>",
"correlationData": "<optional id, copied into the response>",
"configChange": "<updateConfig only: configuration object>",
"params": "<testProducer only: producer connection settings>"
}
]
}
Read the current configuration:
{
"commands": [
{
"command": "getConfig",
"correlationData": "1234"
}
]
}
Replace the configuration:
{
"commands": [
{
"command": "updateConfig",
"correlationData": "12345",
"configChange": {
"eventhubs": [
{
"name": "EvHub1",
"auth": {
"sas": {
"key": "Endpoint=sb://example.servicebus.windows.net/;SharedAccessKeyName=sender;SharedAccessKey=secret;EntityPath=hub1"
}
}
}
],
"producers": [
{
"name": "P1",
"hub": "hub1",
"eventhub": "EvHub1",
"options": {
"messageId": "{NOW_ISO}"
}
}
],
"topicMappings": [
{
"mqttTopics": ["test/topic"],
"target": "P1"
}
]
}
}
]
}
Check that a producer can connect:
{
"commands": [
{
"command": "testProducer",
"correlationData": "1234",
"params": {
"name": "P1",
"hub": "hub1",
"eventhub": {
"name": "EvHub1",
"auth": {
"sas": {
"key": "Endpoint=sb://example.servicebus.windows.net/;SharedAccessKeyName=sender;SharedAccessKey=secret;EntityPath=hub1"
}
}
}
}
}
]
}
A successful testProducer response contains "response": "SUCCESS". Request and response fields are specified in the MQTT API.
JSON schema
Schema for the Azure Event Hubs Bridge. It lists the configuration properties accepted by the plugin.
{
"title": "Azure Bridge Configuration",
"description": "",
"type": "object",
"properties": {
"eventhubs": {
"description": "The Azure EventHubs Bridge can forward messages to multiple EventHubs resources.",
"type": "array",
"items": {
"description": "An Azure EventHubs resource.",
"type": "object",
"properties": {
"name": {
"description": "Unique namespace identifier of an EventHubs",
"type": "string",
"minLength": 1
},
"auth": {
"description": "Namespace authentication. Specify either a SAS connection string or RBAC credentials.",
"type": "object",
"properties": {
"sas": {
"type": "object",
"properties": {
"key": {
"type": "string",
"minLength": 1,
"description": "Event Hubs connection string, including the endpoint."
}
},
"required": [
"key"
]
},
"rbac": {
"type": "object",
"properties": {
"tenantId": {
"type": "string",
"minLength": 1
},
"clientId": {
"type": "string",
"minLength": 1
},
"clientSecret": {
"type": "string",
"minLength": 1
},
"fullyQualifiedNamespace": {
"description": "Full namespace qualifier. If not specified \"<namespace>.servicebus.windows.net\" will be used",
"type": "string",
"minLength": 1
}
},
"required": [
"tenantId",
"clientId",
"clientSecret"
]
}
},
"oneOf": [
{
"required": [
"sas"
]
},
{
"required": [
"rbac"
]
}
]
}
},
"required": [
"name",
"auth"
]
}
},
"producers": {
"description": "Per producer settings for each event hub within an EventHubs resource",
"type": "array",
"items": {
"description": "Producer settings for each event hub within an EventHubs resource",
"type": "object",
"properties": {
"name": {
"description": "Unique name for this producer to use and identify it in mappings",
"type": "string",
"minLength": 1
},
"type": {
"description": "direct sends each event immediately. buffered collects events before sending.",
"type": "string",
"default": "direct",
"enum": [
"direct",
"buffered"
]
},
"hub": {
"description": "Name of the hub to send messages to",
"type": "string",
"minLength": 1
},
"eventhub": {
"description": "Namespace of the EventHubs to use",
"type": "string",
"minLength": 1
},
"auth": {
"description": "Optional credentials merged into the namespace authentication.",
"type": "object",
"properties": {
"sas": {
"type": "object",
"properties": {
"key": {
"type": "string",
"minLength": 1
}
},
"required": [
"key"
]
},
"rbac": {
"type": "object",
"properties": {
"tenantId": {
"type": "string",
"minLength": 1
},
"clientId": {
"type": "string",
"minLength": 1
},
"clientSecret": {
"type": "string",
"minLength": 1
},
"fullyQualifiedNamespace": {
"description": "Full namespace qualifier. If not specified \"<namespace>.servicebus.windows.net\" will be used",
"type": "string",
"minLength": 1
}
},
"required": []
}
},
"oneOf": [
{
"required": [
"sas"
]
},
{
"required": [
"rbac"
]
}
]
},
"options": {
"description": "General producer options",
"type": "object",
"properties": {
"messageId": {
"description": "Value copied to the event message id. Supports {NOW} and {NOW_ISO}.",
"type": "string"
},
"correlationId": {
"description": "Value copied to the event correlation id. Supports {NOW} and {NOW_ISO}.",
"type": "string"
},
"partitionId": {
"description": "Partition id such as \"0\" or \"1\". Do not set partitionKey as well.",
"type": "string"
},
"partitionKey": {
"description": "Partition key. Do not set partitionId as well.",
"type": "object",
"properties": {
"field": {
"description": "JSONPath evaluated against a JSON payload, for example $.device.id.",
"type": "string"
},
"default": {
"description": "Value used when field is set but does not match. Ignored when the payload is not JSON."
}
}
}
},
"anyOf": [
{
"required": [
"partitionId"
],
"not": {
"required": [
"partitionKey"
]
}
},
{
"required": [
"partitionKey"
],
"not": {
"required": [
"partitionId"
]
}
},
{
"not": {
"required": [
"partitionId",
"partitionKey"
]
}
}
]
},
"retryOptions": {
"type": "object",
"properties": {
"mode": {
"type": "string",
"enum": [
"fix",
"exponential"
],
"description": "fix uses a constant delay. exponential increases the delay until maxDelayInMs."
},
"delayInMs": {
"description": "Delay in milliseconds before the next retry.",
"type": "number"
},
"maxDelayInMs": {
"description": "Maximum delay in milliseconds. Used when mode is exponential.",
"type": "number"
},
"maxRetries": {
"description": "Max. number of retries",
"type": "number"
},
"timeoutInMs": {
"description": "Milliseconds to wait before declaring current request as timed out",
"type": "number"
}
}
},
"direct": {
"description": "Options for a direct producer.",
"type": "object",
"properties": {
"maxSizeInBytes": {
"description": "Maximum batch size in bytes. If omitted, Azure Event Hubs selects the size.",
"type": "number"
},
"maxActiveEvents": {
"description": "Maximum number of batches sending or retrying at the same time. Default is 10.",
"type": "number"
}
}
},
"buffered": {
"description": "Options for a buffered producer. Passed through to Azure Event Hubs.",
"type": "object",
"properties": {
"maxWaitTimeInMs": {
"description": "Milliseconds to wait for next message. If none is received during this period the buffered producer sends all of its messages.",
"type": "number"
},
"maxEventBufferLengthPerPartition": {
"description": "Maximum number of events buffered per partition before they are sent. Azure reads this name. Default is 1500.",
"type": "number"
}
}
}
},
"required": [
"name",
"hub",
"eventhub"
]
}
},
"topicMappings": {
"description": "Descriptions for how to map MQTT topics to azure event hubs",
"type": "array",
"items": {
"type": "object",
"properties": {
"mqttTopics": {
"description": "List of MQTT topics to map",
"type": "array",
"items": {
"description": "MQTT topic to map",
"type": "string",
"minLength": 1
}
},
"target": {
"description": "Name of defined producer to send message to",
"type": "string",
"minLength": 1
},
"schemaMapping": {
"description": "Name of an entry in schemaMappings. If omitted, the original MQTT payload is forwarded.",
"type": "string",
"minLength": 1
}
},
"required": [
"mqttTopics",
"target"
]
}
},
"schemaMappings": {
"description": "Optional mappings that build the event body from the MQTT message.",
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {
"type": "string",
"minLength": 1,
"description": "Unique name referenced by topicMappings schemaMapping."
},
"mapping": {
"type": "array",
"description": "Extractors that copy MQTT fields into the outgoing JSON object.",
"items": {
"type": "object",
"properties": {
"source": {
"description": "Metadata or payload path, such as [topic] or [payload][temperature], or a constant.",
"oneOf": [
{
"type": "string"
},
{
"type": "boolean"
},
{
"type": "number"
},
{
"type": "null"
}
]
},
"target": {
"type": "string",
"description": "JSON path in the event body, such as [metadata][topic]."
},
"type": {
"type": "string",
"description": "Optional cast applied to source before it is written. One of datetime, number, integer, uinteger, boolean, string."
},
"options": {
"type": "object",
"description": "Transformations applied to source before it is written. See Extractor options.",
"properties": {
"isConst": {
"type": "boolean",
"description": "Write source as a constant, including when it uses path syntax."
}
},
"additionalProperties": true
}
},
"required": [
"source",
"target"
]
}
}
},
"required": [
"name",
"mapping"
]
}
}
},
"required": [
"eventhubs",
"producers"
]
}