Setting Up Shared Subscriptions

Learn about setting up shared subscriptions in Connectware.

In Connectware, you can use shared subscriptions with mappings and endpoints. Below, we provide three service commissioning file examples that demonstrate different use cases for implementing shared subscriptions.

Syntax of Shared Subscriptions

The syntax for shared subscriptions follows this pattern:

$share/group_name/topic

Parameter	Description	Required
$share	A fixed prefix that designates a shared subscription.	Yes
group_name	An identifier for the group of clients that will share the message load. All clients using the same group name will be part of the same message distribution group.	Yes
topic	The actual topic pattern to subscribe to. Can include wildcards and multi-level topics according to standard MQTT topic rules.	Yes

Example

$share/processing_group/sensors/data

In this example:

• All clients subscribing with this pattern will be part of the processing_group

• Messages published to sensors/data will be distributed among these clients

• Each message will be delivered to exactly one client in the group, preventing duplicate processing

Shared Subscriptions Examples

Basic Example with One Agent

This example allows you to test if shared subscriptions work correctly by forwarding messages published to the input topic to the mapping subscribed to $share/group1/input.

1. Install an agent. See Installing Agents.

2. Register the agent in Connectware. See Registering Agents in Connectware.

3. Apply the service to the agent. See Installing Services.

4. Enable the services. See Enabling Services.

5. Publish a message to the input topic.

Result: The agent will publish to the topic output.

description: |
    Simple Mapping with Shared Subscription

metadata:
    name: Simple Shared Subscription

resources:
    sharedSubscriptionMapping:
        type: Cybus::Mapping
        properties:
            mappings:
                - subscribe:
                      topic: $share/group1/input
                  publish:
                      topic: output
                  rules:
                      - transform:
                            expression: |
                                {
                                  "msg": $
                                }

Advanced Example with Two Agents (Load Balancing)

This example allows you to test the load-balancing capability of shared subscriptions using two agents.

1. Install two agents named agent01 and agent02. See Installing Agents.

2. Register the agents in Connectware. See Registering Agents.

3. Apply the same service to both agents (rename one for differentiation). See Installing Services.

4. Enable the services. See Enabling Services.

5. Publish a message to the input topic.

Result: Either agent01 or agent02 will publish to the output topic, but only one will act at a time. This ensures load balancing between the agents.

description: |
    Shared Subscription Test

metadata:
    name: Shared Subscription Test

parameters:
    agentName:
        description: Choose Agent
        type: string
        enum: ['agent01', 'agent02']

resources:
    sharedSubscriptionMapping:
        type: Cybus::Mapping
        properties:
            agentName: !ref agentName
            mappings:
                - subscribe:
                      topic: $share/group1/input
                  publish:
                      topic: output
                  rules:
                      - transform:
                            expression: !sub |
                                (
                                  $merge([$, {"agent": "${agentName}"}]);
                                )

Advanced Example with Multiple Agents (Wildcard Mappings)

In this example, a shared subscription is deployed across multiple agents. This example sets up a qa group where agents subscribe to the service/# topic wildcard, allowing them to handle messages on all subtopics. This setup is particularly useful for load testing, where tens of thousands of messages per second are processed from sources like OPC UA. The agents then publish these messages to the northbound/opcua/$nodes topic.

The wildcards #nodes and $nodes represent a named wildcard that maps the topic tree under services (e.g., service/a/bc, service/x/y/z/...) to the northbound path (e.g., northbound/opcua/a/bc).

The dedicated connection specified in the subscribe property ensures that each agent creates a new MQTT client instead of sharing a single MqttConnection.

---
description: |
    Wildcard Mapping with Shared Subscription

metadata:
    name: Wildcard Mapping Shared Subscription
    provider: cybus
    homepage: https://www.cybus.io
    version: 1.0.0

parameters:
    agentName:
        type: string
        default: protocol-mapper

    mqttBroker:
        type: string
        default: broker.example.com

    mqttUser:
        type: string
        default: admin

    mqttPassword:
        type: string
        default: admin

resources:
    mqttConnection:
        type: Cybus::Connection
        properties:
            agentName: !ref agentName
            protocol: Mqtt
            targetState: connected
            connection:
                host: !ref mqttBroker
                username: !ref mqttUser
                password: !ref mqttPassword

    wildcardMappingSharedSubscription:
        type: Cybus::Mapping
        properties:
            agentName: !ref agentName
            mappings:
                - subscribe:
                      topic: $share/qa/services/#nodes
                      connection: !ref mqttConnection
                  publish:
                      topic: northbound/opcua/$nodes