FairCom MQ "configureTopic"
JSON MQ "configureTopic" action configures an MQTT topic in FairCom MQ and FairCom Edge
The "configureTopic" action configures an existing MQTT topic or creates a new one. Configuring a topic is similar to altering a topic except the FairCom server automatically creates a topic when you specify a non-existent topic name.
You may send a
"configureTopic"message repeatedly to update as few or many properties as desired. When you omit a property, its current value remains unchanged. Each time you send this message is sent, it modifies the topic’s settings.If the topic has already been published to FairCom's servers,
"configureTopic"configures it.If the topic has not yet been published to FairCom's servers,
"configureTopic"creates and configures it.Creating a topic includes creating an integration table to store the topic’s messages. By default, tables are created in the ctreeSQL database, however, you can reconfigure this to any database you like by changing the configuration or the API session defaults.
Each time this message is sent, it modifies the topic’s settings.
The MQTT Publication service and the MQTT Subscription service are also preconfigured.
Request examples
Minimal request
{
"api": "mq",
"apiVersion": "1.0",
"action": "configureTopic",
"params": {
"topic": "test/topic1"
},
"authToken": "replaceWithAuthTokenFromCreateSession"
}
Note
This example (which shows all of the available parameters) requires additional support in order to work. MQTT messages which clients publish to this topic ("acmefactory/line1/station1/acidbath/telemetry") will be automatically stored in the table's "source_payload" field when they arrive.
The broker will then publish an MQTT message to all subscribers of that topic. The contents of that message will come from the table field specified by the "outputPayloadField" parameter. If that parameter had specified the "source_payload" field, subscribing clients would receive the same message that arrived and was stored in the table. However, in this example, the "outputPayloadField" parameter is specifying the "aValidFieldName" field of the table. That means this topic requires a transform which reads the MQTT message payload from the "source_payload" field, performs some type of transformation on that data, and then stores the results in the "aValidFieldName" field (which will then be copied into the payload of the MQTT messages which the broker sends to all of this topic's subscribers). The "transformName" property specifies the transform which will be used - "acidbath_transform_01" in this example. Before this "configureTopic" action is executed, that transform needs to be created using the "createTransform" action.
If you wanted the MQTT messages to be forwarded to the subscribers as received, you could omit the "outputPayloadField" and "transformName" parameters, which would result in those parameters' default values — "source_payload" and null (meaning "no transform") being used.
{
"api": "mq",
"apiVersion": "1.0",
"requestId": "2",
"authToken": "replaceWithAuthtokenFromCreateSession",
"action": "configureTopic",
"params": {
"topic": "acmefactory/line1/station1/acidbath/telemetry",
"ownerName": "admin",
"databaseName": "faircom",
"tableName": "acmefactory_line1_station1_acidbath_telemetry",
"threads": 1,
"sourcePayloadFormat": "BINARY",
"retentionPolicy": "autoPurge",
"retentionPeriod": 30,
"retentionUnit": "day",
"maxInflightMessages": 20,
"maxDeliveryRatePerSecond": 0,
"retrySeconds": 5,
"downgradeQoS": false,
"newSubscriberDeliveryMode": "newMessages",
"newSubscriberMaxDeliveredMessages": 3,
"outputPayloadField": "aValidFieldName",
"operationIntent": "upsert",
"metadata": {},
"forwardToTopics": [],
"subscribeToExternalBroker": "FALSE",
"transformName": null,
"forwardToExternalBrokers": []
}
}
{
"authToken": "authToken",
"result": {},
"errorCode": 0,
"errorMessage": ""
}
{
"authToken": "authToken",
"result": {},
"requestId": "2",
"errorCode": 0,
"errorMessage": ""
}configureTopic configures an existing MQTT topic or creates a new one
The "params" property is an object that contains an action's parameters. Each action defines its own required and optional properties.
Properties summary
"params" properties summaryProperty | Description | Default | Type | Limits (inclusive) | ||||||
|---|---|---|---|---|---|---|---|---|---|---|
specifies the name of a database | Defaults to the | string | 1 to 64 bytes | |||||||
controls how FairCom Edge manages the Quality of Service (QoS) of subscriptions to this topic |
| Boolean |
| |||||||
forwards all messages sent to this topic to each external broker using the specified topic name |
| array of objects | Zero or more objects | |||||||
| specifies a user-defined name of a connection that allows messages to be forwarded to an external broker |
| string | |||||||
| specifies which quality of service (QoS) to use when forwarding messages to the external broker for the persisted topic |
| integer |
| ||||||
| specifies a unique user-defined name for the topic |
| string | 1 to 65,500 bytes | ||||||
forwards all messages sent to this topic to each of the topics in the array |
| array of strings | One or more strings | |||||||
specifies the throttle rate for current and new subscribers of the topic |
| int32 |
| |||||||
protects subscribers from receiving messages faster than they can handle |
| int32 |
| |||||||
exists primarily for the user interface to find integration information |
| object | May contain any number and type of user-defined properties | |||||||
specifies how new subscribers receive messages |
| string |
| |||||||
specifies the maximum number of previously stored historical messages that a client receives when it subscribes to the topic |
| integer or |
| |||||||
specifies the intent of the operation |
| string |
| |||||||
specifies the field that the MQTT Subscription service should use as the payload when it sends the payload to subscribers |
| string |
| |||||||
specifies the unique name of a schema in a database |
| string | ||||||||
specifies the number of retention units, which controls how long data is retained – see |
| integer |
| |||||||
specifies how messages persist |
| string |
| |||||||
purges expired messages each time this unit cycles – see |
| string |
| |||||||
specifies the number of seconds to wait for the next expected packet during a PUBLISH packet exchange with a client |
| integer |
| |||||||
specifies the variant type format of the |
| string |
| |||||||
connects to the external MQTT broker and subscribes to this topic on that broker |
| string | Any previously defined name of a broker connection (see | |||||||
specifies the name of a table | Optional - No default value | string | 1 to 64 bytes | |||||||
specifies the number of threads assigned to a topic to write published messages to disk |
| integer |
| |||||||
specifies a unique user-defined name for the topic | Required - No default value | string | 1 to 65,500 bytes | |||||||
specifies the name of a transform process. The name cannot be one of the FairCom-provided transform names | Required - No default value | string | 1 to 64 bytes |
The "databaseName" property is an optional string that specifies the database that contains the tables. It defaults to the database name supplied at login.
Note
In the API Explorer, "defaultDatabaseName" is set to "ctreeSQL" in the "createSession" action that happens at login.
A zero-length
"databaseName"is invalid.Its limits are from 0 to 64 bytes.
If the
"databaseName"property is omitted or set tonull, the server will use the default database name specified at login.If no default database is specified during
"createSession","defaultDatabaseName"will be set to the"defaultDatabaseName"value that is specified in theservices.jsonfile.
The "downgradeQoS" property controls how FairCom's servers manage the Quality of Service (QoS) of subscriptions to this topic. This property is optional. The default value is false.
When
false, FairCom's servers honor the QoS requested by each subscriber regardless of the QoS of publishers.When
true, FairCom's servers downgrade the QoS of subscribers to match the QoS of the publisher.
The "forwardToExternalBrokers" property is optional. It can be omitted. It is an array of objects containing one or more brokers to forward the message to. All messages sent to this topic will be forwarded to each of these external brokers using the specified topic name. The default is an empty object.
If there are no items in the list, it is not an error, FairCom's servers simply do not set up any external forwards. This is also how you can configure an existing topic to STOP forwarding to external brokers.
If this broker connection name does not already exist:
FairCom's servers do not set up the forward.
It logs an error stating, "Cannot forward to the external broker because the connection name xxx has not yet been defined by configureBrokerConnection."
A failure does not stop FairCom's servers from trying to set up the next broker in the list.
Each time the
"configureTopic"message is sent, FairCom's servers will completely replace its existing list for"forwardToExternalBrokers"with the new list.Note
Be careful not to create infinite message loops by forwarding topics to each other. For example, topic A forwards to topic B and topic B forwards to topic A. FairCom's servers currently do not detect or shut down infinite message loops.
Example
{
"authToken": "' + token + '",
"api": "mq",
"action": "configureTopic",
"params":
{
"topic": "modbusdata1",
"tableName": "modbusTableTCP",
"forwardToExternalBrokers":
[
{
"brokerConnectionName": "Broker1",
"topic": "modbusdata2",
"forwardQoS": 1
}
]
}
}The "forwardToTopics" property is an array of Topic Names. This property is optional. FairCom's servers will take all messages sent to this topic and automatically forward them to each of the topics in the array. These messages are forwarded within the same instance. The default is an empty array.
If a topic does not already exist, FairCom's servers create it as if an external client sent a message to the topic.
Each time the
"ConfigureTopic"message is sent, FairCom's servers will completely replace its existing list for"forwardToTopics"with the new list.FairCom's servers will not forward a topic to itself. Thus, if the
"forwardToTopics"array contains the topic from"ConfigureTopic", FairCom's servers ignore it.Note
Be careful not to create infinite message loops by forwarding topics to each other. For example, topic A forwards to topic B and topic B forwards to topic A. FairCom's servers currently do not detect or shut down infinite message loops.
This feature is designed to ease the pain of topic migration, which occurs when clients begin publishing to different topics before subscribers can adjust.
The "maxDeliveryRatePerSecond" property sets the throttle rate for current and new subscribers of the topic. This property is optional. If omitted, it defaults to 0, which means no throttling is applied.
It must be a value between
0and2,147,483,647, but a more realistic range is between0and1,000. It is useful to keep a client from being overwhelmed with too many messages sent too quickly.You can set it to a number greater than
0to throttle the delivery rate. This allows you to avoid saturating the network and other resources.For example, you can set it to
100to throttle the delivery to 100 messages per second. This is a maximum setting because the actual message delivery rate is bounded by the network, local resources, the hardware running FairCom's servers, and a client’s ability to process messages quickly.
The "maxInflightMessages" property protects subscribers from receiving messages faster than they can handle. The default value is0. This property is optional.
The default value of
0means there is no maximum limit.This property defines the maximum number of unacknowledged messages that FairCom's servers will send to each subscriber of this topic.
For example, setting this property to
10allows FairCom's servers to send up to 10 publish packets to a subscriber before the servers wait for the subscriber to acknowledge receipt of at least one publish packet.
The "metadata" property is an optional JSON object. It exists primarily for the user interface to find integration information. By default, it is an empty JSON object.
It contains a flexible set of properties.
It typically contains tags and description properties.
Example
{
"description": "",
"tags": [""],
"yourOwnProperties": "usage, purpose, notes, location, etc.",
}
The "newSubscriberDeliveryMode" property defines how new subscribers receive messages. The default is "newMessages".
It may be set to one of two values:
"newMessages""storedMessages"
When set to
"newMessages"new subscribers only receive new messages.When set to
"storedMessages"new subscribers will receive some or all previously sent messages. The"newSubscriberDeliveryMode"property defines the maximum number of messages.
The "newSubscriberMaxDeliveredMessages" property is the maximum number of previously stored historical messages a client receives automatically when subscribing to a topic. The default is 0.
A value of
-1or"all"defaults to delivering all currently stored messages.A non-negative number limits the delivery of currently stored messages to that number, which may be a maximum of 2,147,483,647 messages.
This property only applies when the
"newSubscriberDeliveryMode"property is set to"storedMessages".The broker can only send messages stored in the topic’s integration table. The table's retention policy automatically removes old messages.
Each time a client unsubscribes from a topic and then subscribes to the topic, the broker sends historical messages as specified by this property. A client can use this behavior to reprocess messages, such as refresh a local data cache or recalculate aggregates.
Note
A topic’s "retentionPolicy", "retentionUnit", and "retentionPeriod" properties define how long the broker retains messages in a topic’s integration table.
The "operationIntent" property is optional. It clarifies the intent of the method as an "insert", "update", or "upsert". The default intent is "upsert".
"operationIntent" property values and their descriptions are as follows:"upsert"This is the default value. It treats this method as an
"upsert". If the connection already exists, it updates it. If the connection does not exist, it inserts it. It does not throw an error."insert"It treats this method as an insert. If the connection already exists, FairCom's servers log an error stating, "Cannot insert service connection xxx because it already exists."
"update"It treats this method as an update. If the connection does not already exist, FairCom's servers log an error stating, "Cannot update service connection xxx because it does not exist."
The "outputPayloadField" property specifies the field that the MQTT Subscription service should use as the payload when it sends the payload to subscribers. If omitted or if it is an empty string, it defaults to "source_payload".
"outputPayloadField"must be one of the following values:"source_payload"A user-defined field
This makes it possible for the output from any transform to be used as the payload delivered to subscribers.
The "ownerName" property is an optional string from 1 to 64 bytes that specifies the account that owns an object.
The "retentionPeriod" property specifies how many units of data to retain. It must be an integer value from 1 to 100. It refers to the unit of time specified by the "retentionUnit" property — for example, if "retentionPeriod" is 14 and "retentionUnit" is "day", then message data is retained for 14 days. This property is optional.
If not specified, the default found in the services.json file is used. Initially, it is 4 (weeks).
Automatically purging data is important to ensure that retained data does not consume all storage and shut down the computer. The default value of 4 weeks allows FairCom's servers to store 1 TB of messages when 200 topics send one 2K message per second.
Note
If the value is not an integer from
1to100, FairCom's servers set it to the default value.Smaller numbers improve SQL performance.
Each time the
"retentionPeriod"cycles, FairCom's servers automatically and efficiently delete expired data.FairCom's servers only use the
"retentionPeriod"property when the"retentionPolicy"is"autoPurge".The
"retentionPeriod"can be changed to retain fewer or more messages. Changing it does not necessarily destroy existing data, but data may expire more quickly or be retained longer.The
"retentionPeriod"and"retentionUnit"properties control data granularity as well as the retention time. In other words,"retentionPeriod"defines how many sets of data are stored, and"retentionUnit"defines how often data is purged.For example, if
"rententionPeriod"is set to14, the server stores 14 sets of data. At the beginning of the 15th cycle, the server automatically purges the oldest set of data. If"retentionUnit"is set today, then data will be purged daily. If set to"week", then data will be purged weekly.The current calendar date affects purging.
FairCom's servers automatically purge all retained data that has expired. This is noticeable when FairCom's servers come online after having been offline for a long time. When a server comes back online, it automatically purges all expired data.
For example, if a FairCom server is offline for four weeks when it comes online, it will completely purge all retained data that has a retention time of less than 4 weeks.
The "retentionPolicy" property controls how messages are persisted. This property is optional.
If not specified, the default found in the services.json file is used. Initially, it is "autoPurge".
retentionPolicy values:"autoPurge"This is the default. It is automatically applied when a new topic is created. It is preferred because it allows FairCom's servers to automatically remove messages that are older than the retention time. This helps ensure message data does not consume all storage space. It also minimizes storage costs and speeds up data access. The server partitions a table into multiple files so it can efficiently delete expired files.
"neverPurge"This stores messages on disk and never removes them. This is useful when you need the entire history of the message stream. If message velocity is high, this can consume all storage space and cause an outage. The server creates a non-partitioned table, which is slightly faster than a partitioned table because it stores all records in one file.
"doNotPersist"This stores messages only in RAM, where they are transmitted more quickly. Undelivered messages are lost when FairCom's servers stop and restart. No transforms can be applied to the payload. Message data cannot be bridged across protocols.
Important
Changing the "retentionPolicy" property value to "doNotPersist" removes all existing message history for the topic because it changes how messages are stored on disk.
Each time this unit cycles, FairCom purges expired messages. For example, if you want a week's worth of messages to be purged once a week, set "retentionUnit" to "week" . This property is optional.
If not specified, the default found in the services.json file is used. Initially, it is "week".
This property is used in concert with
"retentionPeriod"to determine retention time."retentionUnit"values:"minute""hour""day""week""month""year""forever"
Note
For best performance, set the
"retentionUnit"to a value that keeps"retentionPeriod"between5and30.When you set
"retentionUnit"property to"forever"the server does not purge messages. This setting is the same as setting"retentionPolicy"to"neverPurge".FairCom MQ only uses the
"retentionUnit"property when the"retentionPolicy"is"autoPurge".
The "retrySeconds" property is the number of seconds that FairCom's servers will wait for the next expected packet during a PUBLISH packet exchange with a client. The default value is 5. It applies to when a client is publishing to the broker and when the broker is publishing to a client. "retrySeconds" is optional.
If this time expires, FairCom's servers will resend the packet to the client and will wait again for the next expected packet.
The minimum value is
1and the maximum value is65535.If
"retrySeconds"is omitted or its range is exceeded, FairCom's servers uses the default value.
The "sourcePayloadFormat" property is an optional string that defines the variant type format of the "source_payload" field. When omitted or set to null, it defaults to "binary".
This property is a string containing the following enumerated values:
"binary""json""utf8""siemensUDT""jpeg""xml"
This property is a hint to the server about the format and type of the MQTT message payload.
This property does not cause server to validate the MQTT payload to see if it matches the type you set. The server stores the payload as is without validation. For example, if you set the type to
"json", it does not stop the server from receiving and storing a non-JSON value or invalid JSON document in the source payload.The FairCom Edge Explorer application may use the value of this property to determine the default way to display a topic's payload.
The transform engine may use the value of this property to help it transform the source payload.
The "subscribeToExternalBroker" property is optional. When this property is set to a valid broker connection, FairCom 's servers connect to the external MQTT broker and subscribes to this topic on that broker. Thus, these servers become a client to another MQTT broker. Each message sent to this topic on the external broker becomes an incoming message in FairCom 's servers – as if it were sent directly to them . The topic works the same as any other topic.
Any client can publish messages to this topic.
A valid Broker Connection must be created before you set
"subscribeToExternalBroker"to a value (see"configureBrokerConnection").If this broker connection name does not already exist, it does not set up an external subscription and it logs an error stating, "Cannot subscribe to the external broker because the Broker Connection name xxx has not yet been defined by ConfigureBrokerConnection."
The "tableName" property is a string containing the name of an integration table that stores the messages of an MQTT topic.
A table name may contain up to 64 ASCII characters and must not start with a number.
You must specify a non-empty string for an integration table name.
FairCom Edge an FairCom MQ automatically create an integration table for each new MQTT topic they process. The first characters in the table name are
"mqtt_msg_"followed by the topic name. Because an MQTT topic can be 65500 bytes and a table name can be 64 bytes, the server may abbreviate the topic name.To better identify the data it holds, rename an integration using the
"tableName"property of the"configureTopic"action.
The "threads" property is the number of threads assigned to a topic to write published messages to disk. This property is optional. The default value is 1.
The default of
1is enough to handle typical workloads.For heavy workloads, you may increase the number of threads to
2or3.
The topic attribute is a unique user-defined name for the topic.
The
"topic"name may be up to 65,500 characters long, but it is typically less than 150 characters.The
"topic"name will be used in all subsequent references to the topic, such as when re-configuring or deleting the topic.
The "transformName" property is an optional string that contains the unique name of a transform process, which consists of one or more transform steps.
A transform is a process that works like a pipeline where the output of one transformation can become the input for another transformation.
The following actions use the
"transformName"property to assign a transform to an integration table:"configureTopic""createInput""alterInput""alterIntegrationTable"