Azure Event Hubs Bridge
Premium
Version 3.1
info
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.
note
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.
note
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 leaves that placeholder as written. Backslashes are then removed.correlationId: event correlation id. The same placeholders asmessageIdare supported.bodyType: type of the event body. Default:string.string: sends the payload text or the JSON text produced by a schema mapping.buffer: sends that same text as UTF-8 bytes. A schema mapping is applied when one is configured.
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.delayInMs: delay in milliseconds before the next attempt. Azure default: 30000. With an exponential delay this is the base interval.maxDelayInMs: upper bound for an exponential delay.timeoutInMs: time in milliseconds before the current attempt is treated as timed out. Azure default: 60000, used when this value is omitted. A number set here is used as given.mode: defines the retry strategy. Usefixfor a fixed retry delay orexponentialto increase retry delay exponentially untilmaxDelayInMsis reached.
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.maxEventBufferLengthPerPartitions: maximum number of events buffered for one partition before they are sent. Azure default: 1500.
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": {
"bodyType": "buffer",
"messageId": "Event sent at {NOW_ISO}"
},
"auth": {
"rbac": {
"fullyQualifiedNamespace": "another_resource_namespace.servicebus.windows.net"
}
},
"retryOptions": {
"maxRetries": 100,
"delayInMs": 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.
The optional schemaMapping defines how the event body is built. Use the name from an entry in schemaMappings array.
[
{
"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 event body is then build using the mapping object with name and mapping if one is present. Following is an example for schemaMappings definition:
[
{
"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 for bodyType string and buffer is a JSON object built from the MQTT message. string then sends that object as JSON text. buffer sends the UTF-8 bytes of that JSON text. Without a schema mapping, string sends the original payload text and buffer sends that text as UTF-8 bytes.
schemaMappings is an optional array. Each entry needs a unique name and a mapping array.
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 matches a metadata path, such as [topic], so it is not read as a path. Text such as [test_data] is already a constant.
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.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 in UTC 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 matches a metadata path, set isConst so the value is not treated as a path:
[
{
"name": "M3",
"mapping": [
{ "source": "[topic]", "target": "[label]", "options": { "isConst": true } },
{ "source": "[payload]", "target": "[payload]" }
]
}
]
This writes the literal string [topic] 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 matches a metadata path, such as[topic]or[payload][temperature]. Other constants, including[test_data], do not need it (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 in UTC 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. If the string does not contain the delimiter, it is left unchanged. 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,joinByis not, andfetchByIndexis not, the slice is joined again with thesplitBydelimiter. If the start index is past the end of the array, the property is omitted (array of two integers).textCase(path:$.schemaMappings[].mapping[].options.textCase):lowercase,uppercase, orcapitalize. Applies to a non-empty string.capitalizeuppercases the first character and lowercases the rest (string).treatNullAsDefault(path:$.schemaMappings[].mapping[].options.treatNullAsDefault): whentrue, a JSONnullis omitted from the event body. Applied beforenullValue,defaultValue,replaceNullWith, andreplaceUndefinedWith(boolean).
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:
replacetruncatesplitBytextCase
replace runs before truncate. With type set to string, both are applied. Otherwise a set replace skips truncate.
{
"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:
replaceNullWithreplaceUndefinedWithnullValuedefaultValueisConsttreatNullAsDefault
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 result is not a number, the original value is kept.integer: parse a string as an integer and drop the fractional part. A comma in a string is treated as a decimal separator before the fractional part is dropped. If the result is not a number, the original value is kept.uinteger: parse a string as an integer, drop the fractional part, and take the absolute value.parseIntstops at a comma. If parsing does not produce a number, the written value isNaN.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, array, or object as a string. Strings are left unchanged, except a string that contains brackets and is not valid JSON, which is quoted again.
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/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"
},
"bodyType": {
"description": "Type of the event body. string (default) sends the payload text or the JSON text produced by a schema mapping. buffer sends that same text as UTF-8 bytes.",
"type": "string",
"enum": ["string", "buffer"],
"default": "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"
},
"maxEventBufferLengthPerPartitions": {
"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 event body is built from the MQTT payload according to bodyType.",
"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"]
}