Skip to content

Commit

Permalink
Adjust shared subscription sample to better fit docs (#406)
Browse files Browse the repository at this point in the history
  • Loading branch information
@TwistedTwigleg
TwistedTwigleg authored Apr 21, 2023
1 parent 4413364 commit 51f3802
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 41 deletions.
10 changes: 5 additions & 5 deletions samples/Mqtt5/SharedSubscription/README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,12 @@ MQTT5 introduces additional features and enhancements that improve the developme

Note: MQTT5 support is currently in **developer preview**. We encourage feedback at all times, but feedback during the preview window is especially valuable in shaping the final product. During the preview period we may make backwards-incompatible changes to the public API, but in general, this is something we will try our best to avoid.

[Shared Subscriptions](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901250) allow IoT devices to connect to a group where messages sent to a topic are then relayed to the group in a round-robin-like fashion. This is useful for distributing message load across multiple subscribing MQTT5 clients automatically. This is helpful for load balancing when you have many messages that need to be processed.
[Shared Subscriptions](https://docs.aws.amazon.com/iot/latest/developerguide/mqtt.html#mqtt5-shared-subscription) allow IoT devices to connect to a group where messages sent to a topic are then relayed to the group in a round-robin-like fashion. This is useful for distributing message load across multiple subscribing MQTT5 clients automatically. This is helpful for load balancing when you have many messages that need to be processed.

Shared Subscriptions rely on a group identifier, which tells the MQTT5 broker/server which IoT devices to treat as a group for message distribution. This is done when subscribing by formatting the subscription topic like the following: `$share/<group identifier>/<topic>`.
Shared Subscriptions rely on a group name/identifier, which tells the MQTT5 broker/server which IoT devices to treat as a group for message distribution. This is done when subscribing by formatting the subscription topic like the following: `$share/<ShareName>/<TopicFilter>`.
* `$share`: Tells the MQTT5 broker/server that the device is subscribing to a Shared Subscription.
* `<group identifier>`: Tells the MQTT5 broker/server which group to add this Shared Subscription to. Messages published to a matching topic will be distributed round-robin amongst the group.
* `<topic>`: The topic that the Shared Subscription is for. Messages published to this topic will be processed in a round-robin fashion. For example, `test/topic`.
* `<ShareName>`: Tells the MQTT5 broker/server which group to add this Shared Subscription to. Messages published to a matching topic will be distributed round-robin amongst the group.
* `<TopicFilter>`: The topic that the Shared Subscription is for. Messages published to this topic will be processed in a round-robin fashion. For example, `test/topic`.

Shared Subscriptions use a round-robbin like method of distributing messages. For example, say you have three MQTT5 clients all subscribed to the same Shared Subscription group and topic. If five messages are sent to the Shared Subscription topic, the messages will likely be delivered in the following order:
* Message 1 -> Client one
Expand Down Expand Up @@ -71,7 +71,7 @@ Replace with the following with the data from your AWS account:
* `<region>`: The AWS IoT Core region where you created your AWS IoT Core thing you wish to use with this sample. For example `us-east-1`.
* `<account>`: Your AWS IoT Core account ID. This is the set of numbers in the top right next to your AWS account name when using the AWS IoT Core website.

Note that in a real application, you may want to avoid the use of wildcards in your ClientID or use them selectively. Please follow best practices when working with AWS on production applications using the SDK. Also, for the purposes of this sample, please make sure your policy allows a client ID of `test-*` to connect or use `--client_id <client ID here>` to send the client ID your policy supports.
Note that in a real application, you may want to avoid the use of wildcards in your ClientID and shared subscription group names/identifiers. Wildcards should only be used selectively. Please follow best practices when working with AWS on production applications using the SDK. Also, for the purposes of this sample, please make sure your policy allows a client ID of `test-*` to connect or use `--client_id <client ID here>` to send the client ID your policy supports.

</details>

Expand Down
64 changes: 28 additions & 36 deletions samples/Mqtt5/SharedSubscription/src/main/java/sharedsubscription/SharedSubscription.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -107,11 +107,9 @@ public void onStopped(Mqtt5Client client, OnStoppedReturn onStoppedReturn) {
*/
static final class SamplePublishEvents implements Mqtt5ClientOptions.PublishEvents {
SampleMqtt5Client sampleClient;
CountDownLatch messagesReceived;

SamplePublishEvents(SampleMqtt5Client client, int messageCount) {
SamplePublishEvents(SampleMqtt5Client client) {
sampleClient = client;
messagesReceived = new CountDownLatch(messageCount);
}

@Override
Expand All @@ -132,7 +130,6 @@ public void onMessageReceived(Mqtt5Client client, PublishReturn publishReturn) {
}
}
}
messagesReceived.countDown();
}
}

Expand All @@ -151,10 +148,10 @@ static final class SampleMqtt5Client {
*/
public static SampleMqtt5Client createMqtt5Client(
String input_endpoint, String input_cert, String input_key, String input_ca,
String input_clientId, int input_count, String input_clientName) {
String input_clientId, String input_clientName) {

SampleMqtt5Client sampleClient = new SampleMqtt5Client();
SamplePublishEvents publishEvents = new SamplePublishEvents(sampleClient, input_count / 2);
SamplePublishEvents publishEvents = new SamplePublishEvents(sampleClient);
SampleLifecycleEvents lifecycleEvents = new SampleLifecycleEvents(sampleClient);

Mqtt5Client client;
Expand Down Expand Up @@ -201,23 +198,17 @@ public static void main(String[] args) {
SampleMqtt5Client subscriberOne = null;
SampleMqtt5Client subscriberTwo = null;

/* Make sure the message count is even */
if (cmdData.input_count%2 != 0) {
onApplicationFailure(new Throwable("'--count' is an odd number. '--count' must be even or zero for this sample."));
System.exit(1);
}

try {
/* Create the MQTT5 clients: one publisher and two subscribers */
publisher = SampleMqtt5Client.createMqtt5Client(
cmdData.input_endpoint, cmdData.input_cert, cmdData.input_key, cmdData.input_ca,
cmdData.input_clientId + '1', cmdData.input_count, "Publisher");
cmdData.input_clientId + '1', "Publisher");
subscriberOne = SampleMqtt5Client.createMqtt5Client(
cmdData.input_endpoint, cmdData.input_cert, cmdData.input_key, cmdData.input_ca,
cmdData.input_clientId + '2', cmdData.input_count, "Subscriber One");
cmdData.input_clientId + '2', "Subscriber One");
subscriberTwo = SampleMqtt5Client.createMqtt5Client(
cmdData.input_endpoint, cmdData.input_cert, cmdData.input_key, cmdData.input_ca,
cmdData.input_clientId + '3', cmdData.input_count, "Subscriber Two");
cmdData.input_clientId + '3', "Subscriber Two");

/* Connect all the clients */
publisher.client.start();
Expand All @@ -231,22 +222,18 @@ public static void main(String[] args) {
System.out.println("[" + subscriberTwo.name + "]: Connected");

/* Subscribe to the shared topic on the two subscribers */
try {
SubscribePacket.SubscribePacketBuilder subscribeBuilder = new SubscribePacket.SubscribePacketBuilder();
subscribeBuilder.withSubscription(input_sharedTopic, QOS.AT_LEAST_ONCE, false, false, SubscribePacket.RetainHandlingType.DONT_SEND);
subscriberOne.client.subscribe(subscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println("[" + subscriberOne.name + "]: Subscribed");
subscriberTwo.client.subscribe(subscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println("[" + subscriberTwo.name + "]: Subscribed");
}
// TMP: If this fails subscribing in CI, just exit the sample gracefully.
catch (Exception ex) {
if (isCI) {
return;
} else {
throw ex;
}
}
SubscribePacket.SubscribePacketBuilder subscribeBuilder = new SubscribePacket.SubscribePacketBuilder();
subscribeBuilder.withSubscription(input_sharedTopic, QOS.AT_LEAST_ONCE, false, false, SubscribePacket.RetainHandlingType.DONT_SEND);
subscriberOne.client.subscribe(subscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println(
"[" + subscriberOne.name + "]: Subscribed to topic '" + cmdData.input_topic +
"' in shared subscription group '" + cmdData.input_groupIdentifier + "'.");
System.out.println("[" + subscriberOne.name + "]: Full subscribed topic is '" + input_sharedTopic + "'.");
subscriberTwo.client.subscribe(subscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println(
"[" + subscriberTwo.name + "]: Subscribed to topic '" + cmdData.input_topic +
"' in shared subscription group '" + cmdData.input_groupIdentifier + "'.");
System.out.println("[" + subscriberTwo.name + "]: Full subscribed topic is '" + input_sharedTopic + "'.");

/* Publish using the publisher client */
PublishPacket.PublishPacketBuilder publishBuilder = new PublishPacket.PublishPacketBuilder();
Expand All @@ -259,9 +246,8 @@ public static void main(String[] args) {
System.out.println("[" + publisher.name + "]: Sent publish");
Thread.sleep(1000);
}
/* Make sure all the messages were gotten on the subscribers */
subscriberOne.publishEvents.messagesReceived.await(60 * 4, TimeUnit.SECONDS);
subscriberTwo.publishEvents.messagesReceived.await(60 * 4, TimeUnit.SECONDS);
/* Wait 5 seconds to let the last publish go out before unsubscribing */
Thread.sleep(5000);
} else {
System.out.println("Skipping publishing messages due to message count being zero...");
}
Expand All @@ -270,9 +256,15 @@ public static void main(String[] args) {
UnsubscribePacket.UnsubscribePacketBuilder unsubscribeBuilder = new UnsubscribePacket.UnsubscribePacketBuilder();
unsubscribeBuilder.withSubscription(input_sharedTopic);
subscriberOne.client.unsubscribe(unsubscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println("[" + subscriberOne.name + "]: Unsubscribed");
System.out.println(
"[" + subscriberOne.name + "]: Unsubscribed to topic '" + cmdData.input_topic +
"' in shared subscription group '" + cmdData.input_groupIdentifier + "'.");
System.out.println("[" + subscriberOne.name + "]: Full unsubscribed topic is '" + input_sharedTopic + "'.");
subscriberTwo.client.unsubscribe(unsubscribeBuilder.build()).get(60, TimeUnit.SECONDS);
System.out.println("[" + subscriberTwo.name + "]: Unsubscribed");
System.out.println(
"[" + subscriberTwo.name + "]: Unsubscribed to topic '" + cmdData.input_topic +
"' in shared subscription group '" + cmdData.input_groupIdentifier + "'.");
System.out.println("[" + subscriberTwo.name + "]: Full unsubscribed topic is '" + input_sharedTopic + "'.");

/* Disconnect all the clients */
publisher.client.stop(null);
Expand Down

0 comments on commit 51f3802

Please sign in to comment.