Follow Broker Enforcements (#240)

Author: pglombardo

Source/HiveMQtt/Client/HiveMQClient.cs

@@ -212,6 +212,28 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message, Cance
     {
         message.Validate();
 
+        // Check if topic alias is used but not supported by broker
+        var topicAliasMaximum = this.Connection?.ConnectionProperties?.TopicAliasMaximum ?? 0;
+        if (message.TopicAlias.HasValue)
+        {
+            if (topicAliasMaximum == 0)
+            {
+                throw new HiveMQttClientException("Topic aliases are not supported by the broker");
+            }
+
+            if (message.TopicAlias.Value > topicAliasMaximum)
+            {
+                throw new HiveMQttClientException($"Topic alias exceeds broker's maximum allowed value of {topicAliasMaximum}");
+            }
+        }
+
+        // Check if retain is used but not supported by broker
+        var retainSupported = this.Connection?.ConnectionProperties?.RetainAvailable ?? true;
+        if (!retainSupported && message.Retain)
+        {
+            throw new HiveMQttClientException("Retained messages are not supported by the broker");
+        }
+
         if (message.QoS.HasValue && this.Connection.ConnectionProperties.MaximumQoS.HasValue &&
             (ushort)message.QoS.Value > this.Connection.ConnectionProperties.MaximumQoS.Value)
         {
@@ -332,6 +354,55 @@ public async Task<SubscribeResult> SubscribeAsync(string topic, QualityOfService
     /// <inheritdoc />
     public async Task<SubscribeResult> SubscribeAsync(SubscribeOptions options)
     {
+        // Check if subscription identifiers are used but not supported by broker
+        var subscriptionIdentifiersSupported = this.Connection?.ConnectionProperties?.SubscriptionIdentifiersAvailable ?? true;
+        if (!subscriptionIdentifiersSupported && options.SubscriptionIdentifier.HasValue)
+        {
+            throw new HiveMQttClientException("Subscription identifiers are not supported by the broker");
+        }
+
+        // Check if retain is used but not supported by broker
+        var retainSupported = this.Connection?.ConnectionProperties?.RetainAvailable ?? true;
+        if (!retainSupported)
+        {
+            // Check if any topic filter has retainAsPublished set to true
+            foreach (var topicFilter in options.TopicFilters)
+            {
+                if (topicFilter.RetainAsPublished is true)
+                {
+                    throw new HiveMQttClientException("Retained messages are not supported by the broker");
+                }
+            }
+        }
+
+        // Check if shared subscriptions are used but not supported by broker
+        var sharedSubscriptionSupported = this.Connection?.ConnectionProperties?.SharedSubscriptionAvailable ?? true;
+        if (!sharedSubscriptionSupported)
+        {
+            // Check if any topic filter contains shared subscription prefix ($share/)
+            foreach (var topicFilter in options.TopicFilters)
+            {
+                if (topicFilter.Topic.StartsWith("$share/", StringComparison.Ordinal))
+                {
+                    throw new HiveMQttClientException("Shared subscriptions are not supported by the broker");
+                }
+            }
+        }
+
+        // Check if wildcards are used but not supported by broker
+        var wildcardSupported = this.Connection?.ConnectionProperties?.WildcardSubscriptionAvailable ?? true;
+        if (!wildcardSupported)
+        {
+            // Check if any topic filter contains wildcards (+ or #)
+            foreach (var topicFilter in options.TopicFilters)
+            {
+                if (topicFilter.Topic.Contains('+') || topicFilter.Topic.Contains('#'))
+                {
+                    throw new HiveMQttClientException("Wildcard subscriptions are not supported by the broker");
+                }
+            }
+        }
+
         // Fire the corresponding event
         this.BeforeSubscribeEventLauncher(options);
 

Tests/HiveMQtt.Test/HiveMQClient/EventExceptionHandlingTest.cs

@@ -23,10 +23,7 @@ public async Task BeforeDisconnectExceptionDoesNotPreventDisconnectAsync()
         Assert.True(client.IsConnected());
 
         // Add event handler that throws an exception
-        client.BeforeDisconnect += (sender, args) =>
-        {
-            throw new Exception("Test exception in BeforeDisconnect");
-        };
+        client.BeforeDisconnect += (sender, args) => throw new InvalidOperationException("Test exception in BeforeDisconnect");
 
         // Attempt to disconnect - should succeed despite the exception
         var disconnectResult = await client.DisconnectAsync().ConfigureAwait(false);
@@ -50,10 +47,7 @@ public async Task OnPublishSentExceptionDoesNotPreventPublishAsync()
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
 
         // Add event handler that throws an exception
-        client.OnPublishSent += (sender, args) =>
-        {
-            throw new Exception("Test exception in OnPublishSent");
-        };
+        client.OnPublishSent += (sender, args) => throw new InvalidOperationException("Test exception in OnPublishSent");
 
         // Attempt to publish - should succeed despite the exception
         var publishResult = await client.PublishAsync(
@@ -85,10 +79,7 @@ public async Task BeforeSubscribeExceptionDoesNotPreventSubscribeAsync()
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
 
         // Add event handler that throws an exception
-        client.BeforeSubscribe += (sender, args) =>
-        {
-            throw new Exception("Test exception in BeforeSubscribe");
-        };
+        client.BeforeSubscribe += (sender, args) => throw new InvalidOperationException("Test exception in BeforeSubscribe");
 
         // Attempt to subscribe - should succeed despite the exception
         var subscribeResult = await client.SubscribeAsync(
@@ -126,10 +117,7 @@ public async Task BeforeUnsubscribeExceptionDoesNotPreventUnsubscribeAsync()
         Assert.NotEmpty(subscribeResult.Subscriptions);
 
         // Add event handler that throws an exception
-        client.BeforeUnsubscribe += (sender, args) =>
-        {
-            throw new Exception("Test exception in BeforeUnsubscribe");
-        };
+        client.BeforeUnsubscribe += (sender, args) => throw new InvalidOperationException("Test exception in BeforeUnsubscribe");
 
         // Attempt to unsubscribe - should succeed despite the exception
         var unsubscribeResult = await client.UnsubscribeAsync(
@@ -167,16 +155,10 @@ public async Task OnMessageReceivedExceptionDoesNotPreventMessageDeliveryAsync()
         Assert.NotEmpty(subscribeResult.Subscriptions);
 
         // Add event handler that throws an exception
-        client.OnMessageReceived += (sender, args) =>
-        {
-            throw new Exception("Test exception in OnMessageReceived");
-        };
+        client.OnMessageReceived += (sender, args) => throw new InvalidOperationException("Test exception in OnMessageReceived");
 
         // Add another event handler to verify message was still delivered
-        client.OnMessageReceived += (sender, args) =>
-        {
-            messageReceived = true;
-        };
+        client.OnMessageReceived += (sender, args) => messageReceived = true;
 
         // Publish a message
         var publishResult = await client.PublishAsync(
@@ -207,10 +189,7 @@ public async Task BeforeConnectExceptionDoesNotPreventConnectAsync()
         var client = new HiveMQClient(options);
 
         // Add event handler that throws an exception
-        client.BeforeConnect += (sender, args) =>
-        {
-            throw new Exception("Test exception in BeforeConnect");
-        };
+        client.BeforeConnect += (sender, args) => throw new InvalidOperationException("Test exception in BeforeConnect");
 
         // Attempt to connect - should succeed despite the exception
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
@@ -238,16 +217,10 @@ public async Task OnPubAckReceivedExceptionDoesNotPreventQoS1DeliveryAsync()
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
 
         // Add event handler that throws an exception
-        client.OnPubAckReceived += (sender, args) =>
-        {
-            throw new Exception("Test exception in OnPubAckReceived");
-        };
+        client.OnPubAckReceived += (sender, args) => throw new InvalidOperationException("Test exception in OnPubAckReceived");
 
         // Add another event handler to verify PubAck was still received
-        client.OnPubAckReceived += (sender, args) =>
-        {
-            pubAckReceived = true;
-        };
+        client.OnPubAckReceived += (sender, args) => pubAckReceived = true;
 
         // Publish a QoS 1 message
         var publishResult = await client.PublishAsync(