Extend PowerShell module with async message handlers and MQTT 5 features

Source/MQTTnet.PowerShell/Cmdlets/ConnectMqttSessionCmdlet.cs

@@ -12,13 +12,15 @@ public class ConnectMqttSessionCmdlet : PSCmdlet
     [Parameter]
     public string? ClientId { get; set; } = Guid.NewGuid().ToString();
 
-    [Parameter(Mandatory = true)]
-    public required new string Host { get; set; }
+    [Parameter(Mandatory = true, ParameterSetName = "Tcp")]
+    [Parameter(Mandatory = true, ParameterSetName = "TcpWithTls")]
+    public new string? Host { get; set; }
 
     [Parameter]
     public string? Password { get; set; }
 
-    [Parameter]
+    [Parameter(ParameterSetName = "Tcp")]
+    [Parameter(ParameterSetName = "TcpWithTls")]
     public int Port { get; set; } = 1883;
 
     [Parameter(Mandatory = true, ValueFromPipeline = true)]
@@ -27,19 +29,90 @@ public class ConnectMqttSessionCmdlet : PSCmdlet
     [Parameter]
     public string? Username { get; set; }
 
-    [Parameter]
+    [Parameter(ParameterSetName = "TcpWithTls")]
     public SwitchParameter UseTls { get; set; }
 
+    [Parameter(ParameterSetName = "TcpWithTls")]
+    public SwitchParameter AllowUntrustedCertificates { get; set; }
+
+    [Parameter(Mandatory = true, ParameterSetName = "WebSocket")]
+    public string? WebSocketUri { get; set; }
+
+    [Parameter]
+    public int KeepAlivePeriod { get; set; } = 15;
+
+    [Parameter]
+    public int Timeout { get; set; } = 10;
+
+    [Parameter]
+    public string? WillTopic { get; set; }
+
+    [Parameter]
+    public string? WillPayload { get; set; }
+
+    [Parameter]
+    public int WillQoS { get; set; } = 0;
+
+    [Parameter]
+    public SwitchParameter WillRetain { get; set; }
+
     protected override void ProcessRecord()
     {
         var clientOptionsBuilder = new MqttClientOptionsBuilder();
-        clientOptionsBuilder.WithTcpServer(Host, Port);
+
+        // Configure transport
+        if (ParameterSetName == "WebSocket")
+        {
+            clientOptionsBuilder.WithWebSocketServer(o => o.WithUri(WebSocketUri!));
+        }
+        else
+        {
+            clientOptionsBuilder.WithTcpServer(Host!, Port);
+
+            if (UseTls)
+            {
+                clientOptionsBuilder.WithTlsOptions(o =>
+                {
+                    if (AllowUntrustedCertificates)
+                    {
+                        o.WithCertificateValidationHandler(_ => true);
+                    }
+                });
+            }
+        }
 
+        // Configure authentication
         if (Username != null)
         {
             clientOptionsBuilder.WithCredentials(Username, Password);
         }
 
+        // Configure session
+        clientOptionsBuilder.WithCleanSession(CleanSession);
+
+        if (!string.IsNullOrEmpty(ClientId))
+        {
+            clientOptionsBuilder.WithClientId(ClientId);
+        }
+
+        // Configure keep alive
+        clientOptionsBuilder.WithKeepAlivePeriod(TimeSpan.FromSeconds(KeepAlivePeriod));
+        clientOptionsBuilder.WithTimeout(TimeSpan.FromSeconds(Timeout));
+
+        // Configure Will message
+        if (!string.IsNullOrEmpty(WillTopic))
+        {
+            clientOptionsBuilder.WithWillTopic(WillTopic);
+
+            if (!string.IsNullOrEmpty(WillPayload))
+            {
+                clientOptionsBuilder.WithWillPayload(System.Text.Encoding.UTF8.GetBytes(WillPayload));
+            }
+
+            clientOptionsBuilder.WithWillQualityOfServiceLevel((Protocol.MqttQualityOfServiceLevel)WillQoS);
+            clientOptionsBuilder.WithWillRetain(WillRetain);
+        }
+
         var response = Session.GetClient().ConnectAsync(clientOptionsBuilder.Build()).GetAwaiter().GetResult();
 
         WriteObject(response);

Source/MQTTnet.PowerShell/Cmdlets/GetMqttSessionStatusCmdlet.cs

@@ -0,0 +1,27 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Management.Automation;
+
+namespace MQTTnet.PowerShell.Cmdlets;
+
+[Cmdlet(VerbsCommon.Get, "MqttSessionStatus")]
+[OutputType(typeof(MqttClientConnectionStatus))]
+public class GetMqttSessionStatusCmdlet : PSCmdlet
+{
+    [Parameter(Mandatory = true, ValueFromPipeline = true)]
+    public required PsMqttSession Session { get; set; }
+
+    protected override void ProcessRecord()
+    {
+        var client = Session.GetClient();
+        var status = new
+        {
+            IsConnected = client.IsConnected,
+            Options = client.Options
+        };
+
+        WriteObject(status);
+    }
+}

Source/MQTTnet.PowerShell/Cmdlets/PublishMqttMessageCmdlet.cs

@@ -1,6 +1,8 @@
+using System.Collections;
 using System.Management.Automation;
 using System.Text;
 using MQTTnet.Protocol;
+using MQTTnet.Packets;
 
 namespace MQTTnet.PowerShell.Cmdlets;
 
@@ -22,29 +24,44 @@ public class PublishMqttMessageCmdlet : PSCmdlet
     [Parameter(Mandatory = true)]
     public required string Topic { get; set; }
 
+    [Parameter]
     public string? ContentType { get; set; }
 
+    [Parameter]
     public string? ResponseTopic { get; set; }
 
+    [Parameter]
     public ushort TopicAlias { get; set; }
 
+    [Parameter]
     public uint MessageExpiryInterval { get; set; }
 
+    [Parameter]
+    public Hashtable? UserProperties { get; set; }
+
     protected override void ProcessRecord()
     {
-        // if (Session == null || !Session.IsConnected)
-        //     throw new InvalidOperationException("Session not connected.");
-
-        var msg = new MqttApplicationMessageBuilder().WithTopic(Topic)
+        var msgBuilder = new MqttApplicationMessageBuilder()
+            .WithTopic(Topic)
             .WithPayload(Encoding.UTF8.GetBytes(Payload ?? string.Empty))
             .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
             .WithRetainFlag(Retain)
             .WithContentType(ContentType)
             .WithResponseTopic(ResponseTopic)
             .WithTopicAlias(TopicAlias)
-            .WithMessageExpiryInterval(MessageExpiryInterval)
-            .Build();
+            .WithMessageExpiryInterval(MessageExpiryInterval);
+
+        // Add user properties if provided
+        if (UserProperties != null && UserProperties.Count > 0)
+        {
+            foreach (var key in UserProperties.Keys)
+            {
+                var value = UserProperties[key]?.ToString() ?? string.Empty;
+                msgBuilder.WithUserProperty(key.ToString(), value);
+            }
+        }
 
+        var msg = msgBuilder.Build();
         var response = Session.GetClient().PublishAsync(msg).GetAwaiter().GetResult();
 
         WriteObject(response);

Source/MQTTnet.PowerShell/Cmdlets/RegisterMqttMessageHandlerCmdlet.cs

@@ -14,10 +14,43 @@ public class RegisterMqttMessageHandlerCmdlet : PSCmdlet
 
     protected override void ProcessRecord()
     {
-        EventHandler<PsMqttMessage> handler = (s, e) =>
+        // Capture the ScriptBlock and current default runspace (available during cmdlet execution)
+        var scriptBlock = Action ?? throw new ArgumentNullException(nameof(Action));
+        var capturedRunspace = Runspace.DefaultRunspace ?? throw new InvalidOperationException("No default runspace available during cmdlet execution.");
+        var capturedHost = Host;
+
+        EventHandler<PsMqttMessage> handler = (sender, e) =>
         {
-            //throw new NotImplementedException();
-            //InvokeCommand.InvokeScript(Action, false, PipelineResultTypes.Output, null, new object[] { e.Topic, e.Payload });
+            // Run asynchronously to avoid blocking the event thread
+            Task.Run(() =>
+            {
+                // Temporarily set the default runspace for this thread
+                var originalRunspace = Runspace.DefaultRunspace;
+                Runspace.DefaultRunspace = capturedRunspace;
+
+                try
+                {
+                    // Invoke the ScriptBlock directly with arguments (returns Collection<PSObject>)
+                    var results = scriptBlock.Invoke(e.Topic, e.Payload, e); // Pass topic, payload, and full message
+
+                    // Handle output manually (e.g., write to host/console)
+                    foreach (var result in results)
+                    {
+                        capturedHost.UI.WriteLine(result?.ToString() ?? string.Empty);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    // Handle errors by writing to host error stream
+                    capturedHost.UI.WriteErrorLine($"MQTT handler error: {ex.Message}");
+                    // Optionally: capturedHost.UI.WriteVerboseLine(ex.StackTrace); for more details
+                }
+                finally
+                {
+                    // Restore the original default runspace for this thread
+                    Runspace.DefaultRunspace = originalRunspace;
+                }
+            });
         };
 
         Session.MessageReceived += handler;

Source/MQTTnet.PowerShell/Cmdlets/ShowMqttExamplesCmdlet.cs

@@ -0,0 +1,78 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Management.Automation;
+
+namespace MQTTnet.PowerShell.Cmdlets;
+
+[Cmdlet(VerbsCommon.Show, "MqttExamples")]
+public class ShowMqttExamplesCmdlet : PSCmdlet
+{
+    protected override void ProcessRecord()
+    {
+        var examples = @"
+MQTTnet PowerShell Module - Quick Examples
+==========================================
+
+1. Basic Connection and Publish:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
+   Publish-MqttMessage -Session $session -Topic 'test/topic' -Payload 'Hello'
+   Disconnect-MqttSession -Session $session
+   Remove-MqttSession -Session $session
+
+2. Subscribe and Receive Messages:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
+   Subscribe-MqttTopic -Session $session -Topic 'test/demo'
+   $msg = Receive-MqttMessage -Session $session -TimeoutSeconds 30
+   Write-Host ""Received: $($msg.Payload)""
+
+3. WebSocket Connection:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -WebSocketUri 'ws://broker.hivemq.com:8000/mqtt'
+   Publish-MqttMessage -Session $session -Topic 'test/ws' -Payload 'Hello WS'
+
+4. TLS/SSL Connection:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -Host 'broker.hivemq.com' -Port 8883 `
+       -UseTls -AllowUntrustedCertificates
+
+5. Async Message Handler:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
+   Subscribe-MqttTopic -Session $session -Topic 'test/#'
+   Register-MqttMessageHandler -Session $session -Action {
+       param($topic, $payload, $message)
+       Write-Host ""Message on $topic : $payload""
+   }
+
+6. Last Will and Testament:
+   $session = New-MqttSession
+   Connect-MqttSession -Session $session -Host 'broker.hivemq.com' `
+       -WillTopic 'status/client' -WillPayload 'Offline' -WillQoS 1 -WillRetain
+
+7. Publish with User Properties:
+   $props = @{ 'source' = 'PowerShell'; 'version' = '1.0' }
+   Publish-MqttMessage -Session $session -Topic 'test/props' `
+       -Payload 'Data' -UserProperties $props
+
+8. High QoS Publishing:
+   Publish-MqttMessage -Session $session -Topic 'critical/data' `
+       -Payload 'Important' -QoS 2
+
+9. Test Connection:
+   $isAlive = Test-MqttConnection -Session $session
+   Write-Host ""Connection alive: $isAlive""
+
+10. Get Session Status:
+    $status = Get-MqttSessionStatus -Session $session
+    Write-Host ""Connected: $($status.IsConnected)""
+
+For more details, see: Source/MQTTnet.PowerShell/README.md
+";
+
+        WriteObject(examples);
+    }
+}

Source/MQTTnet.PowerShell/Cmdlets/TestMqttConnectionCmdlet.cs

@@ -0,0 +1,41 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Management.Automation;
+
+namespace MQTTnet.PowerShell.Cmdlets;
+
+[Cmdlet(VerbsDiagnostic.Test, "MqttConnection")]
+[OutputType(typeof(bool))]
+public class TestMqttConnectionCmdlet : PSCmdlet
+{
+    [Parameter(Mandatory = true, ValueFromPipeline = true)]
+    public required PsMqttSession Session { get; set; }
+
+    [Parameter]
+    public int TimeoutSeconds { get; set; } = 5;
+
+    protected override void ProcessRecord()
+    {
+        try
+        {
+            var client = Session.GetClient();
+
+            if (!client.IsConnected)
+            {
+                WriteObject(false);
+                return;
+            }
+
+            var cts = new CancellationTokenSource(TimeSpan.FromSeconds(TimeoutSeconds));
+            client.PingAsync(cts.Token).GetAwaiter().GetResult();
+
+            WriteObject(true);
+        }
+        catch
+        {
+            WriteObject(false);
+        }
+    }
+}

Source/MQTTnet.PowerShell/Cmdlets/UnregisterMqttMessageHandlerCmdlet.cs

@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Management.Automation;
+
+namespace MQTTnet.PowerShell.Cmdlets;
+
+[Cmdlet(VerbsLifecycle.Unregister, "MqttMessageHandler")]
+public class UnregisterMqttMessageHandlerCmdlet : PSCmdlet
+{
+    [Parameter(Mandatory = true, ValueFromPipeline = true)]
+    public required PsMqttSession Session { get; set; }
+
+    protected override void ProcessRecord()
+    {
+        // Clear all event handlers
+        Session.ClearMessageHandlers();
+        WriteObject($"All message handlers cleared for MQTT session.");
+    }
+}