Actions

In HiveMQ Data Hub, an action is a collection of operations the broker performs in response to the outcome of a data-validation or a state transition in behavior-validation.
Actions give you the option to specify what happens once a validation finishes or a client transitions through various states to which the configured behavior model applies.

The available functions vary based on the type of the policy.
For more information, see Actions in Data Policies and Actions in Behavior Policies.

Actions in Data Policies

The outcome of each MQTT message validation in the validations portion of a data policy can be success or failure.

Based on the validation outcome, you can define what the HiveMQ broker does in optional onSuccess and onFailure actions. The onSuccess and onFailure actions each contain a pipeline of operations that executes tasks in the defined order to achieve the desired result.

Table 1. Available onSuccess and onFailure parameters
Parameter Description

pipeline

Lists a sequence of operations in an array. HiveMQ executes the operations in the order they are listed.
Each operation contains the following information:

  • id: The user-defined string that identifies the operation. For example, log-my-log-message.

  • functionId: The string that identifies the type of function you want the HiveMQ broker to execute. For example, Delivery.redirectTo.

  • arguments: Defines the necessary parameters of the function referenced in the functionId field of the onSuccess or onFailure pipeline.
    For more information, see non-terminal-functions and terminal-functions.

Example onSuccess and onFailure configurations in a data policy
  "onSuccess": {
    "pipeline": [
      {
        "id": "logSuccess",
        "functionId": "System.log",
        "arguments": {
          "level": "INFO",
          "message": "${clientId} sent a valid publish on topic '${topic}' with result '${validationResult}'"
        }
      }
    ]
  },
  "onFailure": {
    "pipeline": [
      {
        "id": "logFailure",
        "functionId": "System.log",
        "arguments": {
          "level": "WARN",
          "message": "${clientId} sent an invalid publish on topic '${topic}' with result '${validationResult}'"
        }
      }
    ]
  }

The onSuccess and onFailure action pipelines in a data policy have different default behaviors:

  • onSuccess: If there is no terminal function in the pipeline, the MQTT message is published to the original topic. The MQTT message is acknowledged according to the MQTT specification.

  • onFailure: If there is no terminal function in the pipeline, the MQTT message is published to the original topic. However, actions such as dropping the message can be defined.
    The publishing MQTT client is handled as follows:

    • MQTT 3: The client is disconnected.

    • MQTT 5: A PUBACK with the reason string The publish processing was prevented by a policy. and the reason code 131 (Implementation Specific Error) is sent.

For more information, see Functions.

Transformations in Data Policies

The Data Hub policy engine can execute transformation functions in the action pipelines of a data policy.
All transformation functions are non-terminal functions. For each MQTT PUBLISH packet the data policy processes and references, the policy engine invokes the configured transformation function with the MQTT publish and a Context object.
The resulting publish object is returned by the script and passed to the next function in a pipeline. Multiple transformation functions can be executed sequentially.

For more information, see Transformations.

Transformation Syntax in Data Policies

The following syntax can be used to reference a transformation function in an action pipeline of a data policy:

  • fn:function-id:latest where fn is a necessary prefix to identify a script, function-id specifies the function identifier during script creation, and latest states the latest version of the script. If desired, you can also specify a particular script version.
    For more information on versioning, see Create Script. For more information on transformation functions, see Non-terminal Functions.

Deserialization & Serialization

Incoming and outgoing MQTT message payloads can be any arbitrary sequence of bytes.
However, in various application scenarios it can be useful to execute an intermediate transformation step as a script to ensure data is suitable for its intended purpose.

Every transformation script requires two steps:

  1. A deserialization step to make incoming data accessible

  2. A serialization step to transmit a byte sequence to consumers

Transformation functions can be embedded into your action pipelines. Functions are passed in a publish-object. The publish object includes a deserialized payload that requires a deserialization. The deserialization step converts incoming MQTT payloads into an internal JSON representation. The JSON object is made available to the scripts for standardized data manipulation, regardless of the incoming data representation.

For example, for JavaScript, we provide JavaScript-flavored object manipulation.

A corresponding serialization step is required to serialize the transformed data into the desired data format.

The following structure is executed:

  • The incoming MQTT payload is deserialized according to a specified schema using a deserialization function. For more information, see Serdes.deserialize.

  • Transformation scripts are executed step by step. Multiple functions can be executed sequentially.

  • Finally, the transformed data is serialized to publish to the actual topic. For more information, see Serdes.serialize.

Example transformation in an action pipeline

The following example shows a transformation script with the prerequisite deserialization and serialization:

Example action pipeline
"onSuccess": {
    "pipeline": [
      {
        "id": "operation-2eng0",
        "functionId": "Serdes.deserialize",
        "arguments": {
          "schemaId": "schema-from-sensor"
          "schemaVersion": "latest",
        }
      },
      {
        "id": "operation-ek2Mx",
        "functionId": "fn:fahrenheit-to-celsius:latest",
        "arguments": {}
      },
      {
        "id": "operation-4DBF3",
        "functionId": "Serdes.serialize",
        "arguments": {
          "schemaId": "schema-for-fan"
          "schemaVersion": "latest",
        }
      }
    ]
}

The first action in this pipeline configures a Serdes.deserialize function.
This function deserializes the incoming MQTT message according to the schema-from-sensor schema.
Next, the fahrenheit-to-celsius transformation function is executed to perform a unit conversion.
Finally, to complete the action pipeline, the Serdes.serialize function serializes the transformed message into JSON format according to a new schema-for-fan schema.

Actions in Behavior Policies

HiveMQ Data Hub behavior policies can model the journey of an MQTT client from the initial state, through intermediate states, to a final success or failed terminal state.
Each transition the client makes from one state to another state includes an event that uniquely specifies the to state and from state.
For example, an OnInboundConnect event in a client transitions from an initial state to a connected state.

In a behavior policy, the transitions an MQTT client makes from one state to the next in the HiveMQ broker in response to an event can be used to trigger an optional pipeline of actions.

The action pipeline contains one or more operations that HiveMQ performs in the configured order. For example, an action pipeline that contains functions to print a log entry and increment a specific metric. For more information, see Functions.

Transitions in Behavior Policies

In the onTransition configuration, you can precisely identify a particular transition in the referenced behavior model with a fromState and`toState` along with an event. Depending on the type of event, an action pipeline comprising a set of available functions can tbe triggered.
For more information, see Available behavior policy events and Functions.

Table 2. Available behavior policy events
Mqtt.OnInboundConnect The event that occurs when the MQTT client connects to the broker with an MQTT CONNECT packet.

Mqtt.OnInboundPublish

The event that occurs when the MQTT client publishes an MQTT PUBLISH packet.

Mqtt.OnInboundSubscribe

The event that occurs when the MQTT client sends a MQTT SUBSCRIBE packet.

Mqtt.OnInboundDisconnect

The event that occurs when the MQTT client sends an MQTT DISCONNECT packet.

Connection.OnDisconnect

The event that occurs when the TCP connection of the MQTT client closes.

Functions

You can use two categories of functions in your policies. Both categories of functions can be executed on any outcome.

  • non-terminal functions: Allow further operations in the pipeline to be executed.
    For example, the System.log function logs a message to the hivemq.log file and allows further steps to be executed.

  • terminal functions: End further operations in the pipeline. The first terminal function in a pipeline stops the execution.
    For example, the Delivery.redirectTo function publishes an MQTT message to a certain topic and stops further executions.

Non-terminal Functions

Non-terminal functions manipulate data or initiate additional activity without terminating the action pipeline.

Currently, the following non-terminal functions are supported:

System.log (data and behavior policies)

Logs a message on the given level.

Table 3. Available System.log function arguments
Argument Type Values Description

level

String

DEBUG, ERROR, WARN, INFO, TRACE

Specifies the log level of the function in the hivemq.log file.

message

String

Adds a user-defined string that prints to the log file. For more information, see Example log message.

Example user-defined System.log function arguments
{
  "id": "log-my-log-message",
  "functionId": "System.log",
  "arguments": {
    "level": "INFO",
    "message": "My defined log message"
  }
}

To learn more about how the System.log function can be used in a policy, see the full running example in our HiveMQ policy cookbooks repository.

Metrics.Counter.increment (data and behavior policies)

Increments a metric of type counter, which can be accessed with monitoring. If the metric does not exist yet, it will be created by the first execution of the function.

Table 4. Available Metrics.Counter.increment function arguments
Argument Type Values Description

metricName

String

Specifies the name of the metric to be incremented.

NOTE: Interpolation is currently not supported for this argument.

incrementBy

Number

Specifies the amount by which the counter should be incremented. Negative values are supported in order to support decrement operation.

For HiveMQ Enterprise Edition, the created metric has the prefix com.hivemq.data-hub.custom.counters. For HiveMQ Edge, the created metric has the prefix com.hivemq.edge.data-hub.custom.counters.
Example user-defined Metrics.Counter.increment function arguments
{
  "id": "increment-my-counter",
  "functionId": "Metrics.Counter.increment",
  "arguments": {
    "metricName": "my-counter",
    "incrementBy": 1
  }
}

To learn more about how the Metrics.Counter.increment function can be used in a policy, see the full running example in our HiveMQ policy cookbooks repository.

Mqtt.UserProperties.add (data and behavior policies)

Adds a user property to the MQTT message. The Mqtt.UserProperties.add function is available for both data and behavior policies.

Table 5. Available Mqtt.UserProperties.add function arguments
Argument Type Values Description

name

String

Specifies the name of the user property. Note that multiple user properties with the same name are allowed.

value

String

Specifies the value of the user property.

Example user-defined Mqtt.UserProperties.add function arguments
{
  "id": "add-my-user-property",
  "functionId": "Mqtt.UserProperties.add",
  "arguments": {
    "name": "my-user-property",
    "value": "my-value"
  }
}

To learn more about how the Mqtt.UserProperties.add function can be used in a policy, see the full running example in our HiveMQ policy cookbooks repository.

Serdes.deserialize (data policies)

Deserializes a binary MQTT message payload into a data object based on the configured JSON Schema or Protobuf schema.

Table 6. Available Serdes.deserialize function arguments
Argument Type Values Description

schemaId

String

The identifier of the JSON Schema or Protobuf schema to be used for deserialization.

schemaVersion

String

The version of the schema to be used for deserialization.

Example
{
  "id": "my-deserialization",
  "functionId": "Serdes.deserialize",
  "arguments": {
    "schemaId": "schema1",
    "schemaVersion": "latest"
  }
}

Serdes.serialize (data policies)

Serializes a data object into a binary MQTT message payload based on the configured JSON Schema or Protobuf schema.

Table 7. Available Serdes.serialize function arguments
Argument Type Values Description

schemaId

String

The identifier of the JSON Shema to be used for serialization. Support for serialization with a Protobuf schema is not yet available.

schemaVersion

String

The version of the schema to be used for serialization.

Example
{
  "id": "my-serialization",
  "functionId": "Serdes.serialize",
  "arguments": {
    "schemaId": "schema1",
    "schemaVersion": "latest"
  }
}

Terminal Functions

Terminal functions manipulate data or initiate additional activity and terminate the pipeline of the action.
When an operation in a pipeline contains a terminal function, no further operations after that operation are executed.
Additionally, no further matching policies are evaluated.

Currently, the following terminal functions are supported. Some functions are limited to specific types of policies:

Delivery.redirectTo (data policies only)

Redirects an MQTT PUBLISH message to a specified topic.
The Delivery.redirectTo function is available for data policies only.

Table 8. Available Delivery.redirectTo function arguments
Argument Type Values Description

topic

String

The destination MQTT topic according to MQTT specification.

applyPolicies

Boolean

true, false

Defines whether policies are executed after publishing to a different topic. Possible values are true and false.
NOTE: If multiple policies have a terminal action Delivery.redirectTo with applyPolicies set to true, HiveMQ evaluates a maximum of 20 policies. To avoid endless loops, no additional policies are evaluated once the limit is reached.

Example user-defined Delivery.redirectTo function arguments
{
  "id": "redirect-to-operation",
  "functionId": "Delivery.redirectTo",
  "arguments": {
    "topic": "redirected/my-topic",
    "applyPolicies": false
  }
}

To learn more about how the Delivery.redirectTo function can be used in a policy, see the full running example in our HiveMQ policy cookbooks repository.

Mqtt.disconnect (data and behavior policies)

Disconnects an MQTT client. The Mqtt.disconnect function is available for both data and behavior policies. The Mqtt.disconnect function has no arguments.

Example Mqtt.disconnect
{
  "id": "client-disconnect",
  "functionId": "MQTT.disconnect",
  "arguments": {}
}

Mqtt.drop (data and behavior policies)

Drops the MQTT packet that is currently processed. The Mqtt.drop function is available for data and behavior policies.

In a behavior policy, the Mqtt.drop function is suitable for the Mqtt.OnIncomingPublish and Mqtt.OnIncomingSubscribe events only. The Mqtt.drop function is not supported for DISCONNECT packets. For CONNECT packets, use the Mqtt.disconnect function instead.

The Mqtt.drop function has no arguments.

Example Mqtt.drop
{
  "id": "drop-publish",
  "functionId": "Mqtt.drop",
  "arguments": {}
}

String Interpolation

You can interpolate function arguments that have the type string.
For example, invalid_messages/${topic} interpolates to invalid_message/sensor_data for the variable topic set to sensor_data.
In this case, variables that begin with a dollar sign are with curly brackets ${} are interpolated during policy execution.

Interpolation enhances your ability to design flexible topic redirects, custom log messages, and more.

Table 9. Available predefined variables
Variable Type Description Example Value Data Policy Behavior Policy

clientId

String

The MQTT client ID

a6dc66f2-efec-45e8-922a-3ee1ee7a68d7

topic

String

The MQTT topic to which the MQTT message was published

myhome/groundfloor/livingroom

policyId

String

The id of the policy that is currently executed

com.hivemq.policy.coordinates

validationResult

String

A textual description of the validation result. This text can contain schema validation errors for further debugging.

ValidationResults:[{schemaId=schema1, success=true}]

fromState

String

Textual representation of the state of the state machine before the transition.

Initial

toState

String

Textual representation of the state to which the state machine transitions.

Connected

triggerEvent

String

Textual representation of the event that triggered the state machine transition.

MQTT - Inbound PUBLISH

timestamp

Long

Current time in milliseconds since the UNIX epoch (Jan 1, 1970).

1696925620296